#include "config.h"

#include <unistd.h>
#include <string.h>
#include <stdlib.h>
#include <stdio.h>
#include <getopt.h>
#include <dirent.h>
#include <ctype.h>
#include <libgen.h>

#define DBG_SUBSYS S_LIBINTERFACE

#include "configure.h"
#include "env.h"
#include "adt.h"
#include "net_table.h"
#include "lichstor.h"
#include "system.h"
#include "cluster.h"
#include "metadata.h"
#include "lich_md.h"
#include "storage.h"
#include "volume.h"
#include "lichbd.h"
#include "license.h"
#include "md_map.h"
#include "net_global.h"
#include "utils.h"
#include "dbg.h"
#include "system.h"
#include "lsv_bitmap.h"
#include "../../storage/controller/table_proto.h"
#include "../../storage/storage/stor_rpc.h"
#include "utils.h"
#include "cJSON.h"
#include "hostmap.h"

int object_md5sum(const fileid_t *oid);

#define MULTITHREADING          1
#define COPY_THREAD_MAX         10
#define COPY_SIZE_MIN   (4*1024*1024)  /* 4M */

#define GROUPFLAG ""                    /* prefix for group snapshot */
#define GROUP(new, old) sprintf(new, "%s%s", GROUPFLAG, old)

#define SNAPSHOT_SHOW_JSON 1
#define SNAPSHOT_SHOW_TREE 2

#define STOR_STAT_STEP_MAX (1000 * 50) /*50g*/
#define STOR_STAT_THREAD_MAX (2)

typedef struct {
        int fd;
        lichbd_ioctx_t ioctx;
        fileid_t snapid;
        void * (*worker)(void *_arg);
} arg_ext_t;

typedef struct {
        sem_t sem;
        int ret;
        arg_ext_t ext;
        off_t offset;
        size_t size;
} arg_t;

typedef struct {
        fileid_t fileid;
        off_t begin;
        off_t end;
        int fill;
        int user_pipeline;
        int pipeline;
        int64_t used;
        int ret;
        uint64_t success;
        char pool[MAX_NAME_LEN];
        pthread_t th;
} allocate_arg_t;

typedef struct {
        allocate_arg_t *arg;
        sem_t sem;
        sem_t stop;
        int running;
        int thread;
        uint64_t total;
} allocate_arg_count_t;

typedef struct {
        int idx;
        nid_t nid;
        fileid_t fileid;
        off_t off;
        size_t size;
        int ret;
        filestat_t filestat;
} stor_stat_t;

int __is_all_digit(const char *_str)
{
        int i, len = strlen(_str);

        for (i = 0; i < len; i++) {
                if (_str[i] < '0' || _str[i] > '9')
                        return 0;
        }

        return 1;
}

int utils_get_size(const char *_str, uint64_t *_size)
{
        int ret;
        uint64_t size = 0, base;
        char unit, str[MAX_NAME_LEN];

        YASSERT(_str);

        if (strlen(_str) < 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        strcpy(str, _str);
        unit = str[strlen(str) - 1];

        if (unit == 'i') {
                base = 1024;
                unit = str[strlen(str) - 2];
                str[strlen(str) - 2] = 0;
        } else {
                base = 1000;
                str[strlen(str) - 1] = 0;
        }

        if (!__is_all_digit(str)) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        size = atoll(str);

        switch (unit) {
        case 'b':
        case 'B':
                break;
        case 'k':
        case 'K':
                size *= base;
                break;
        case 'm':
        case 'M':
                size *= (base * base);
                break;
        case 'g':
        case 'G':
                size *= (base * base * base);
                break;
        case 't':
        case 'T':
                size *= (base * base * base * base);
                break;
        default:
                fprintf(stderr, "size unit must be specified, see help for detail\n");
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (size <= 0 || size % LICH_BLOCK_SPLIT != 0) {
                fprintf(stderr, "size must be greater than zero and aligned by %d!\n", LICH_BLOCK_SPLIT);
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        *_size = size;

        return 0;
err_ret:
        return ret;
}

int utils_mkdir(const char *pool, const char *dir, int flag, ec_t *ec, const char *_storage_area)
{
        int ret;
        char name[MAX_NAME_LEN], buf[MAX_NAME_LEN];
        fileid_t fid, rootid;

        /*
           if (utils_spare_policy(pool)) {
           DERROR("reach the spare policy\n");
           ret = ENOSPC;
           GOTO(err_ret, ret);
           }
           */

        if (!strncmp(dir, ISCSI_ROOT, strlen(ISCSI_ROOT))) {
                ret = _valid_iscsi_path(dir);
                if (ret) {
                        ret = EINVAL;
                        GOTO(err_ret, ret);
                }
        }

        ret = md_root(&rootid, pool);
        if (ret)
                GOTO(err_ret, ret);

        ret = stor_splitpath(pool, dir, &fid, name);
        if (unlikely(ret)) {
                if (ret == ENOENT && flag) {
                        strcpy(buf, dir);

                        ret = utils_mkdir(pool, dirname(buf), 1, ec, _storage_area);
                        if (ret) {
                                GOTO(err_ret, ret);
                        }

                        ret = stor_splitpath(pool, dir, &fid, name);
                        if (ret) {
                                GOTO(err_ret, ret);
                        }
                } else
                        GOTO(err_ret, ret);
        }

        ret = stor_mkpool_with_area(&fid, name, ec, _storage_area, NULL, NULL);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_rmdir(const char *pool, const char *dir)
{
        int ret;
        char name[MAX_NAME_LEN];
        fileid_t fid;

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, dir, &fid, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_rmpool(pool, &fid, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_touch_with_area(const char *pool, const char *file, vol_param_t *param)
{
        int ret;
        char name[MAX_NAME_LEN];
        fileid_t parentid, fid;

        /*
           if (utils_spare_policy(pool)) {
           DERROR("reach the spare policy\n");
           ret = ENOSPC;
           GOTO(err_ret, ret);
           }
           */

        if (!strncmp(file, ISCSI_ROOT, strlen(ISCSI_ROOT))) {
                ret = _valid_iscsi_path(file);
                if (ret) {
                        ret = EINVAL;
                        GOTO(err_ret, ret);
                }
        }

        ret = stor_splitpath(pool, file, &parentid, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        strcpy(param->name, name);

        ret = stor_mkvol_with_area(&fid, &parentid, param);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_touch(const char *pool, const char *file, int priority)
{
        vol_param_t param;

        vol_param_init(&param);
        param.priority = priority;

        return utils_touch_with_area(pool, file, &param);
}

int utils_rmvol(const char *pool, const char *file, int force)
{
        int ret;
        char name[MAX_NAME_LEN];
        fileid_t parent, fid;

        DBUG("pool %s path %s force %d\n", pool, file, force);

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, file, &parent, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &parent, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (fid.type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (!force) {
                ret = md_check_connection(&fid);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        if (force) {
                ret = utils_volume_has_protected_snapshot(pool, file);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        ret = stor_rmvol(&parent, name, force);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = unmap_hosts_by_fileid(id2str(&fid));
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_statstr(const char *pool, const fileid_t *parent, const char *name, char *statstr)
{
        int ret;
        char stat[128], perm[128];
        fileid_t fid;
        struct stat stbuf;

        ret = stor_lookup(pool, parent, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_getattr(pool, &fid, &stbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        stat_to_datestr(&stbuf, stat);
        mode_to_permstr(stbuf.st_mode, perm);

        sprintf(statstr, "%s %d %d %d %ju %s %s",
                        perm, 1, stbuf.st_gid, stbuf.st_uid, stbuf.st_size, stat, name);

        return 0;
err_ret:
        return ret;
}

void utils_statstr_error(const char *name, char *statstr)
{
        sprintf(statstr, "?????????? ? ? ? ? ??? ?? ??:?? %s", name);
}

static int __utils_list_print(void *pool, void *_fid, void *_name, void *_flag)
{
        int ret, *flag = _flag;
        fileid_t *fid = _fid;
        char statstr[MAX_PATH_LEN], *name = _name;

        memset(statstr, 0x00, MAX_PATH_LEN);
        ret = utils_statstr(pool, fid, name, statstr);
        if (unlikely(ret)) {
                utils_statstr_error(name, statstr);
        }

        switch (*flag) {
        case 0:
                if (statstr[0] == '-') {
                        printf("%s\n", statstr);
                }
                break;
        case 1:
                if (statstr[0] == 'd') {
                        printf("%s\n", statstr);
                }
                break;
        default:
                printf("%s\n", statstr);
                break;
        }

        return 0;
}

static int __utils_list_print_json(void *pool, void *_parent, void *_name, void *_flag)
{
        int ret;
        fileid_t *parent = _parent;
        fileid_t fid;
        char *name = _name;
        struct stat stbuf;
        char date[128], perm[128], type[128];

        ret = stor_lookup(pool, parent, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_getattr(pool, &fid, &stbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        stat_to_datestr(&stbuf, date);
        mode_to_permstr(stbuf.st_mode, perm);

        if (S_ISDIR(stbuf.st_mode))
                strcpy(type, "directory");
        else if (S_ISREG(stbuf.st_mode))
                strcpy(type, "regular file");
        else
                strcpy(type, "unknown");

        cJSON *array = _flag;
        cJSON *obj = cJSON_CreateObject();
        cJSON_AddItemToArray(array, obj);
        cJSON_AddStringToObject(obj, "File", name);
        cJSON_AddStringToObject(obj, "Access", perm);
        cJSON_AddStringToObject(obj, "Type", type);
        cJSON_AddNumberToObject(obj, "Uid", stbuf.st_uid);
        cJSON_AddNumberToObject(obj, "Gid", stbuf.st_gid);
        cJSON_AddNumberToObject(obj, "Size", stbuf.st_size);
        cJSON_AddStringToObject(obj, "Birth", date);

        return 0;
err_ret:
        return 0;
}

/**
 * flag:  0 -- only show file
 *        1 -- only show dir
 *        other -- show all
 */
int utils_list(const char *pool, const char *dir, int flag, int output_format)
{
        int ret;
        cJSON *json = NULL, *array = NULL;
        char realPath[MAX_PATH_LEN];

        switch(output_format) {
        case DEFAULT_FORMAT:
                ret = utils_iterator(pool, dir, __utils_list_print, &flag);
                break;
        case JSON_FORMAT:
                json = cJSON_CreateObject();
                array = cJSON_CreateArray();
                sprintf(realPath, "%s%s", pool, dir);
                cJSON_AddItemToObject(json, realPath, array);
                ret = utils_iterator(pool, dir, __utils_list_print_json, (void *)array);
                printf("%s\n", cJSON_PrintUnformatted(json));
                cJSON_Delete(json);
                break;
        default:
                fprintf(stderr, "Don't support output format.\n");
                ret = EINVAL;
                break;
        }

        return ret;
}

int utils_iterator(const char *pool, const char *dir, func_int3_t func, void *arg)
{
        int ret, done = 0;
        uint64_t offset = 0, offset2 = 0;
        fileid_t fid;
        int delen;
        struct dirent *de0, *de;
        char uuid[MAX_NAME_LEN] = {};

        ret = stor_lookup1(pool, dir, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        get_uuid(uuid);
        ret = stor_listpool_open(&fid, uuid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        de0 = NULL;
        while (done == 0) {
                ret = stor_listpool(&fid, uuid, offset, (void **)&de0, &delen);
                if (unlikely(ret)) {
                        GOTO(err_close, ret);
                }

                if (delen == 0) {
                        break;
                }

                offset2 = 0;
                dir_for_each(de0, delen, de, offset2) {
                        if (strlen(de->d_name) == 0) {
                                done = 1;
                                break;
                        } else if (delen - (int)offset2 < (int)(sizeof(*de) + MAX_NAME_LEN)) {
                                break;
                        }

                        offset += de->d_reclen;

                        if (!strcmp(de->d_name, ".") || !strcmp(de->d_name, "..")) {
                                continue;
                        }

                        ret = func((void *)pool, &fid, de->d_name, arg);
                        if (unlikely(ret))
                                GOTO(err_close, ret);

                }

                yfree((void **)&de0);
        }

        if (de0)
                yfree((void **)&de0);

        stor_listpool_close(&fid, uuid);

        return 0;
err_close:
        stor_listpool_close(&fid, uuid);
err_ret:
        return ret;
}

int utils_find_list(const char *pool, const char *parent, const char *key, int maxlevel, int level)
{
        int ret, retry = 0;
        static int flag = 0;
        off_t offset = 0, offset2 = 0;
        fileid_t fid;
        char stat[128], perm[128];
        char name[MAX_NAME_LEN];
        struct stat stbuf;
        int delen, done = 0;
        struct dirent *de0, *de;
        char uuid[MAX_NAME_LEN] = {};

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

retry1:
        ret = stor_lookup1(pool, parent, &fid);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        ret = stor_getattr(pool, &fid, &stbuf);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        stat_to_datestr(&stbuf, stat);
        mode_to_permstr(stbuf.st_mode, perm);

        /*
         *printf("%s %d %d %d %ju %s --- %s\n",
         *                perm, 1, stbuf.st_gid, stbuf.st_uid, stbuf.st_size, stat, parent);
         */

        strcpy(name, parent);
        if (key == NULL || !strcmp(key, basename(name))) {
                flag++;
                printf("/%s%s\n", pool, parent);
        }

        if (S_ISDIR(stbuf.st_mode)) {
                get_uuid(uuid);

        retry2:
                ret = stor_listpool_open(&fid, uuid);
                if (unlikely(ret)) {
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_ret, ret, retry2, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_ret, ret);
                }

                de0 = NULL;
                while (done == 0) {
                retry3:
                        ret = stor_listpool(&fid, uuid, offset, (void **)&de0, &delen);
                        if (unlikely(ret)) {
                                if (ret == EAGAIN) {
                                        USLEEP_RETRY(err_close, ret, retry3, retry, 50, (100 * 1000));
                                } else
                                        GOTO(err_close, ret);
                        }

                        if (delen == 0)
                                break;

                        offset2 = 0;

                        dir_for_each(de0, delen, de, offset2) {
                                if (strlen(de->d_name) == 0 || strcmp(de->d_name, "/") == 0) {
                                        done = 1;
                                        break;
                                } else if (delen - (int) offset2 < (int) (sizeof(*de) + MAX_NAME_LEN))
                                        break;

                                offset += de->d_reclen;

                                if (!strcmp(de->d_name, ".") || !strcmp(de->d_name, "..")) {
                                        continue;
                                }

                                if (!strcmp(parent, "/")) {
                                        sprintf(name, "%s%s", parent, de->d_name);
                                } else {
                                        sprintf(name, "%s/%s", parent, de->d_name);
                                }

                                if (maxlevel == 0 || level < maxlevel) {
                                        ret = utils_find_list(pool, name, key, maxlevel, level + 1);
                                        if (unlikely(ret))
                                                GOTO(err_close, ret);
                                }
                        }

                        yfree((void **)&de0);
                }

                if (de0)
                        yfree((void **)&de0);

                stor_listpool_close(&fid, uuid);
        }

        if (flag == 0) {
                ret = ENOENT;
                GOTO(err_ret,ret);
        }
        return 0;
err_close:
        stor_listpool_close(&fid, uuid);
err_ret:
#if 0
        strcpy(name, parent);
        if (key == NULL || !strcmp(key, basename(name))) {
                printf("%s ???\n", parent);
        }
#endif
        return ret;
}

int utils_find(const char *pool, const char *path, const char *name, int maxlevel)
{
        int ret;

        ret = utils_find_list(pool, path, name, maxlevel, 0);
        if (unlikely(ret)) {
                if (name) {
                        printf("/%s%s/%s\n", pool, path, name);
                } else {
                        printf("/%s%s\n", pool, path);
                }
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_rename(const char *pool, const char *from, const char *to)
{
        int ret;
        fileid_t fromdir, todir;
        char fromname[MAX_NAME_LEN], toname[MAX_NAME_LEN];

        if (!from || !to) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (!strncmp(to, ISCSI_ROOT, strlen(ISCSI_ROOT))) {
                ret = _valid_iscsi_path(to);
                if (ret) {
                        ret = EINVAL;
                        GOTO(err_ret, ret);
                }
        }

        ret = stor_splitpath(pool, from, &fromdir, fromname);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, to, &todir, toname);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_rename(pool, &fromdir, fromname, &todir, toname);
        if (unlikely(ret)) {
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

/**
 * format:   0  -- lich
 *           1  -- json
 *           2  -- xml
 */
int utils_stat(const char *pool, const char *path, int format)
{
        int ret;
        fileid_t fid;
        struct stat stbuf;
        char date[32], perm[32], type[32];
        char _path[MAX_PATH_LEN];
        char realPath[MAX_PATH_LEN];

        path_normalize(path,_path);

        ret = stor_lookup1(pool, _path, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_getattr(pool, &fid, &stbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        stat_to_datestr(&stbuf, date);
        mode_to_permstr(stbuf.st_mode, perm);

        if (S_ISDIR(stbuf.st_mode))
                strcpy(type, "directory");
        else if (S_ISREG(stbuf.st_mode))
                strcpy(type, "regular file");
        else
                sprintf(type, "unknown(%x)", stbuf.st_mode);

        sprintf(realPath, "%s%s", pool, _path);
        if (format == 1) {
                 cJSON *json = cJSON_CreateObject();
                 cJSON_AddStringToObject(json, "File", realPath);
                 cJSON_AddNumberToObject(json, "Size", stbuf.st_size);
                 cJSON_AddStringToObject(json, "Type", type);
                 cJSON_AddStringToObject(json, "Access", perm);
                 cJSON_AddNumberToObject(json, "Uid", stbuf.st_uid);
                 cJSON_AddNumberToObject(json, "Gid", stbuf.st_gid);
                 cJSON_AddStringToObject(json, "Access", date);
                 cJSON_AddStringToObject(json, "Modify", date);
                 cJSON_AddStringToObject(json, "Change", date);
                 cJSON_AddStringToObject(json, "Birth", date);
                 printf("%s\n", cJSON_PrintUnformatted(json));
                 cJSON_Delete(json);
        } else {
                printf("  File: '%s' Id: "CHKID_FORMAT"\n"
                                "  Size: %ju %s\n"
                                "Access: (%o/%s) Uid: (%u) Gid: (%u)\n"
                                "Access: %s\n"
                                "Modify: %s\n"
                                "Change: %s\n"
                                " Birth: %s\n",
                                realPath, CHKID_ARG(&fid),
                                stbuf.st_size, type,
                                stbuf.st_mode, perm, stbuf.st_uid, stbuf.st_gid,
                                date,
                                date,
                                date,
                                date);
        }

        return 0;
err_ret:
        return ret;
}

int utils_truncate(const char *pool, const char *path, size_t size)
{
        int ret;
        fileid_t fileid;

        /*
           if (utils_spare_policy(pool)) {
           DERROR("reach the spare policy\n");
           ret = ENOSPC;
           GOTO(err_ret, ret);
           }
           */

        ret = stor_lookup1(pool, path, &fileid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_truncate(pool, &fileid, size);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_cat(const char *pool, const char *path, uint64_t offset, __off_t length)
{
        int ret;
        fileid_t parentid, fid;
        char name[MAX_NAME_LEN], buf[MAX_BUF_LEN + 1];
        struct stat stbuf;
        __off_t left;
        uint64_t size;
        lichbd_ioctx_t ioctx;
        int localize = 0;

        ret = stor_splitpath(pool, path, &parentid, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &parentid, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_getattr(pool, &fid, &stbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = lichbd_connect(pool, path, &ioctx, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        left = length > 0 ? length : stbuf.st_size;
        while (left > 0) {
                size = left < MAX_BUF_LEN ? left : MAX_BUF_LEN;

                ret = lichbd_pread1(&ioctx, buf, size, offset, localize);
                if (ret < 0) {
                        ret = -ret;
                        GOTO(err_ret, ret);
                }

                YASSERT((uint64_t)ret == size);
                buf[size] = '\0';
                printf("%s", buf);
                fflush(stdout);

                offset += size;
                left -= size;
        }

        return 0;
err_ret:
        return ret;
}

int utils_write(const char *pool, const char *value, const char *path, uint64_t offset)
{
        int ret;
        fileid_t parentid, fid;
        char name[MAX_NAME_LEN];
        lichbd_ioctx_t ioctx;

        ret = stor_splitpath(pool, path, &parentid, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_mkvol(&parentid, name, NULL, &fid);
        if (unlikely(ret)) {
                if (ret != EEXIST)
                        GOTO(err_ret, ret);
        }

        ret = lichbd_connect(pool, path, &ioctx, O_CREAT);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = lichbd_pwrite1(&ioctx, value, strlen(value) + 1, offset, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_fsstat(const char *pool)
{
        int ret;
        fileid_t fid;
        struct statvfs svbuf;

        ret = stor_lookup1(pool, "/", &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_statvfs(pool, &fid, &svbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        printf("bsize:%ld\nblocks:%ld\nbused:%ld\nbfree:%ld\n",
                        svbuf.f_bsize, svbuf.f_blocks, svbuf.f_blocks - svbuf.f_bavail, svbuf.f_bfree);
        return 0;
err_ret:
        return ret;
}

int __snapshot_copy_thread_create(arg_t * arg)
{
        int ret;
        pthread_t th;
        pthread_attr_t ta;

        ret = sem_init(&arg->sem, 0, 0);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        (void) pthread_attr_init(&ta);
        (void) pthread_attr_setdetachstate(&ta, PTHREAD_CREATE_DETACHED);

        ret = pthread_create(&th, &ta, arg->ext.worker, arg);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int __snapshot_copy_thread_startup(uint64_t _left, arg_ext_t *arg_ext)
{
        int ret, count, i, err = 0;
        uint64_t left, offset, size, avg_size;
        arg_t args[COPY_THREAD_MAX];

        count = 0;
        offset = 0;
        left = _left;
        avg_size = _ceil(left, COPY_THREAD_MAX);
        _align(&avg_size, NULL, COPY_SIZE_MIN);

        while (left > 0) {
                size = left < avg_size ? left : avg_size;
                args[count].offset = offset;
                args[count].size = size;
                args[count].ret = 0;
                args[count].ext = *arg_ext;

                ret = __snapshot_copy_thread_create(&args[count]);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                count ++;
                offset += size;
                left -= size;
        }

        for (i = 0; i < count; i++) {
                ret = sem_wait(&args[i].sem);
                if (unlikely(ret))
                        err = ret;

                ret = args[i].ret;
                if (unlikely(ret))
                        err = ret;
        }

        if (err) {
                ret = err;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __snapshot_list_one(const char *pool, const char *name, char *buf, int *buflen, fileinfo_t *fileinfo)
{
        int ret;
        chkid_t chkid;
        char uuid[MAX_NAME_LEN] = {};

        get_uuid(uuid);
        ret = stor_lookup1(pool, name, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (chkid.type == __POOL_CHUNK__) {
                ret = EISDIR;
                GOTO(err_ret, ret);
        }

        ret = md_getattr(&chkid, fileinfo);
        if (ret)
                GOTO(err_ret, ret);

        ret = stor_snapshot_list(&chkid, uuid, 0, buf, buflen);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __is_snapshot_exist(const char *pool, const char *vol, const char *snap)
{
        int ret;
        char buf[BIG_BUF_LEN];
        int buflen = BIG_BUF_LEN;
        struct dirent *de;
        uint64_t offset = 0;
        char snap_name[MAX_NAME_LEN];
        long snap_ver, snap_from;
        fileinfo_t fileinfo;

        ret = __snapshot_list_one(pool, vol, buf, &buflen, &fileinfo);
        if (ret)
                GOTO(err_ret, ret);

        dir_for_each(buf, buflen, de, offset) {
                if (strlen(de->d_name) == 0) {
                        break;
                }

                ret = sscanf(de->d_name, "%s %ld %ld", snap_name, &snap_ver, &snap_from);
                if (ret != 3)
                        continue;

                if (!strcmp(snap, snap_name)) {
                        return 1;
                }
        }

        return 0;
err_ret:
        return -ret;
}

static int __snapshot_check(const char *pool, const char *vol, const char *snap)
{
        int ret;
        chkid_t chkid;

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (chkid.type == __POOL_CHUNK__) {
                ret = EISDIR;
                GOTO(err_ret, ret);
        }

        ret = stor_snapshot_check(&chkid, snap);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = 0;
err_ret:
        return ret;
}

inline static int __group_snapshot_split(const char *src, char *des, char *list[], int *count)
{
        int ret;

        *count = MAX_GROUP_VOLUMES_NUM;
        strcpy(des, src);
        _str_split(des, ',', list, count);
        if (*count < 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __group_snapshot_create(const char *pool, const char *vols, const char *_snap)
{
        int ret, count, i;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];
        chkid_t chkids[MAX_GROUP_VOLUMES_NUM];
        const char *_site[MAX_GROUP_VOLUMES_NUM];
        char buf[MAX_BUF_LEN], *curr;

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        curr = buf;
        *curr = '\0';
        for (i = 0; i < count; i++) {
#if 0
                ret = __is_snapshot_exist(pool, list[i], snap);
                if (ret) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = EEXIST;
                        GOTO(err_ret, ret);
                }

                ret = __snapshot_check(list[i], snap);
                if (ret)
                        GOTO(err_ret, ret);
#endif

                ret = stor_lookup1(pool, list[i], &chkids[i]);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }

                ret = md_chunk_getsite(pool, &chkids[i], curr);
                if (ret)
                        GOTO(err_ret, ret);

                _site[i] = curr;
                curr += strlen(curr) + 1;
                YASSERT(curr - buf <= MAX_BUF_LEN);
        }

        ret = stor_group_snapshot_create(chkids, snap, count, _site);
        if (ret)
                GOTO(err_ret, ret);

        printf("create success\n");
        return 0;
err_ret:
        return ret;
}

static int __snapshot_create(const char *pool, const char *vol, const char *snap, int p)
{
        int ret;
        chkid_t chkid;
        fileinfo_t fileinfo;
        char _site[MAX_NAME_LEN];

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (chkid.type != __VOLUME_CHUNK__) {
                ret = EISDIR;
                GOTO(err_ret, ret);
        }

        ret = md_chunk_getsite(pool, &chkid, _site);
        if (ret)
                GOTO(err_ret, ret);

        ret = md_getattr(&chkid, &fileinfo);
        if (ret) {
                GOTO(err_ret, ret);
        }

        ret = stor_snapshot_create(&chkid, snap, p, _site);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __snapshot_rollback(const char *pool, const char *vol, const char *snap);
int utils_snapshot_create(const char *pool, const char *name, int force, int p)
{
        int ret, count, name_len;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        /*
           if (utils_spare_policy(pool)) {
           DERROR("reach the spare policy\n");
           ret = ENOSPC;
           GOTO(err_ret, ret);
           }
           */

        name_len = strlen(name);
        if (name_len > MAX_GROUP_NAME_LEN) {
                fprintf(stderr, "(groups + snapshot) len should less than (%d), your input (%d)\n",
                                MAX_GROUP_NAME_LEN - 1, name_len);
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        name_len = strlen(list[1]);
        if (name_len > MAX_NAME_LEN) {
                fprintf(stderr, "snapshot len should less than (%d), your input (%d)\n",
                                MAX_NAME_LEN - 1, name_len);
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (!force && !is_valid_name(list[1], "snap")) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                if(strcmp(list[1], "test100000") == 0)
                {
                        int i;

                        for(i=0;i<100000;i++)
                        {
                                char name[32];

                                sprintf(name, "snap%d", i);

                                ret = __snapshot_create(pool, list[0], name, p);
                                if( ((i+1) % 20) == 0)
                                {
                                        __snapshot_rollback(pool, list[0], "snap1");
                                }
                        }
                }
                else
                        ret = __snapshot_create(pool, list[0], list[1], p);
        } else {
                ret = __group_snapshot_create(pool, list[0], list[1]);
        }

        if (ret) {
                if (ret == EEXIST) {
                        fprintf(stderr, "WARN: snapshot %s exists (%d)\n", name, ret);
                } else if (ret == EDQUOT) {
                        fprintf(stderr, "WARN: Maximum support for %d snapshots. Currently %d.\n",
                                TABLE_PROTO_USER_SNAP_MAX, TABLE_PROTO_USER_SNAP_MAX + 1);
                        ret = EPERM;
                } else {
                        fprintf(stderr, "WARN: snapshot %s %s (%d)\n", name, strerror(ret), ret);
                }
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __snapshot_rollback(const char *pool, const char *vol, const char *snap)
{
        int ret;
        chkid_t chkid;

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_snapshot_rollback(&chkid, snap);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __group_snapshot_rollback(const char *pool, const char *vols, const char *_snap)
{
        int ret, count, i, failed;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __is_snapshot_exist(pool, list[i], snap);
                if (ret != 1) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = ENOENT;
                        GOTO(err_ret, ret);
                }

                ret = __snapshot_check(pool, list[i], snap);
                if (ret)
                        GOTO(err_ret, ret);
        }

        failed = 0;
        for (i = 0; i < count; i++) {
                ret = __snapshot_rollback(pool, list[i], snap);
                if (ret) {
                        failed++;
                        printf("rollback %s@%s failed\n", list[i], _snap);
                }
        }

        if (failed) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        printf("rollback successed\n");
        return 0;
err_ret:
        return ret;
}

static int __snapshot_rollback_inspection(const char *pool, const char *vol)
{
        int ret;
        char buf[BIG_BUF_LEN], *group;
        int buflen = BIG_BUF_LEN;
        struct dirent *de;
        uint64_t offset = 0;
        char snap_name[MAX_NAME_LEN];
        long snap_ver, snap_from;
        fileinfo_t fileinfo;

        ret = __snapshot_list_one(pool, vol, buf, &buflen, &fileinfo);
        if (ret)
                GOTO(err_ret, ret);

        dir_for_each(buf, buflen, de, offset) {
                if (strlen(de->d_name) == 0) {
                        break;
                }

                if (!strcmp(de->d_name, ".") || !strcmp(de->d_name, "..")) {
                        continue;
                }

                ret = sscanf(de->d_name, "%s %ld %ld", snap_name, &snap_ver, &snap_from);
                if (ret != 3)
                        continue;

                group = strstr(snap_name, GROUPFLAG);
                if (group && strlen(group) == strlen(snap_name))
                        goto err_ret;
        }

        return 0;
err_ret:
        return 1;
}

int utils_snapshot_rollback(const char *pool, const char *name)
{
        int ret, count;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                if (strlen(GROUPFLAG)) {
                        ret = __snapshot_rollback_inspection(pool, list[0]);
                        if (ret) {
                                ret = EPERM;
                                GOTO(err_ret, ret);
                        }
                }

                ret = __snapshot_rollback(pool, list[0], list[1]);

        } else {
                ret = __group_snapshot_rollback(pool, list[0], list[1]);
        }
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

static int __snapshot_remove(const char *pool, const char *vol, const char *snap, int force)
{
        int ret;
        chkid_t chkid;

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_snapshot_remove(pool, &chkid, snap, force);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return abs(ret);

}

static int __group_snapshot_remove(const char *pool, const char *vols, const char *_snap)
{
        int ret, count, i, failed = 0;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __snapshot_remove(pool, list[i], snap, 0);
                if (ret) {
                        failed++;
                        printf("remove %s@%s failed, %s\n", list[i], _snap, strerror(ret));
                } else {
                        printf("remove %s@%s success\n", list[i], _snap);
                }
        }

        if (failed) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_remove(const char *pool, const char *name, int force)
{
        int ret, count;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                ret = __snapshot_remove(pool, list[0], list[1], force);
        } else {
                ret = __group_snapshot_remove(pool, list[0], list[1]);
        }
        if (ret)
                GOTO(err_ret, ret);

        //printf("remove success\n");

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_src(const char *pool, const char *file, const char *snap, fileid_t *fileid,
                fileid_t *snapid, struct stat *stbuf, int is_lsv)
{
        int ret;
        fileid_t *src_fileid;

        if (is_lsv) {
                src_fileid = fileid;
        } else {
                ret = stor_lookup1(pool, file, fileid);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }

                ret = stor_lookup(pool, fileid, snap, snapid);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }

                src_fileid = snapid;
        }

        ret = stor_getattr(pool, src_fileid, stbuf);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

#if 0
static int __utils_snapshot_clone_table(const fileid_t *snapid, const fileid_t *fileid)
{
        int ret;
        buffer_t buf;

        ret = mbuffer_init(&buf, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = md_chunk_table_read(snapid, snapid, &buf, TABLE_PROTO_XATTR_AREA, TABLE_PROTO_XATTR);
        if (unlikely(ret)) {
                GOTO(err_free, ret);
        }

        ret = md_chunk_table_write(fileid, fileid, &buf, TABLE_PROTO_XATTR_AREA, TABLE_PROTO_XATTR);
        if (unlikely(ret)) {
                GOTO(err_free, ret);
        }

        mbuffer_pop(&buf, NULL, buf.len);

        ret = md_chunk_table_read(snapid, snapid, &buf, TABLE_PROTO_XMAP_AREA, TABLE_PROTO_XMAP);
        if (unlikely(ret)) {
                GOTO(err_free, ret);
        }

        ret = md_chunk_table_write(fileid, fileid, &buf, TABLE_PROTO_XMAP_AREA, TABLE_PROTO_XMAP);
        if (unlikely(ret)) {
                GOTO(err_free, ret);
        }

        mbuffer_free(&buf);

        return 0;
err_free:
        mbuffer_free(&buf);
err_ret:
        return ret;
}

#else

static int __utils_snapshot_clone_table(const fileid_t *snapid, const fileid_t *fileid)
{
        int ret, buflen = BIG_BUF_LEN, count, retry, i;
        char buf[BIG_BUF_LEN], *list[100], key[MAX_NAME_LEN], value[MAX_NAME_LEN];

        ret = md_xattr_list(NULL, snapid, buf, &buflen);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        count = 100;
        _str_split(buf, '\n', list, &count);

        for (i = 0; i < count; i++) {
                if (strlen(list[i]) == 0)
                    break;
                
                ret = sscanf(list[i], "%[^:]: %[^\n]", key, value);
                YASSERT(ret == 2);

                DWARN("set "CHKID_FORMAT" key %s value %u\n", CHKID_ARG(fileid), key, value);
                retry = 0;
        retry:
                ret = md_xattr_set(NULL, fileid, key, value, strlen(value) + 1, 0);
                if (unlikely(ret)) {
                        ret = _errno(ret);
                        USLEEP_RETRY(err_ret, ret, retry, retry, 100, (100 * 1000));
                }
        }

        return 0;
err_ret:
        return ret;
}

#endif

int __utils_snapshot_clone(const char *dest_pool, snap_clone_param_t *clone_param)
        // const fileid_t *fileid, const fileid_t *snapid,
        //                              fileinfo_t *fileinfo, struct stat *stbuf,
        //                              const char *to, int p, const char *_site)
{
        int ret, retry = 0;
        fileid_t *old_fileid, parent, newfile;
        char name[MAX_NAME_LEN], src[MAX_NAME_LEN];
        setattr_t setattr;
        vol_param_t param;

retry0:
        ret = stor_splitpath(dest_pool, clone_param->to_vol_name, &parent, name);
        if (unlikely(ret)) {
                ret = _errno(ret);
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        memset(&setattr, 0x0, sizeof(setattr));

        if (IS_RICH_LSV(clone_param->volume_format)) {
                old_fileid = &clone_param->vol_fileid;
                md_info2attr((void *)&setattr, &clone_param->vol_fileinfo);
        } else {
                old_fileid = &clone_param->snap_fileid;
                md_info2attr((void *)&setattr, &clone_param->snap_fileinfo);
        }
#if LSV
        setattr.size.set_it = SET_LOGICAL_SIZE;
#else
        setattr.size.set_it = 1;
#endif
        setattr.size.size = clone_param->stbuf.st_size;

#if LSV
        if (IS_RICH_LSV(clone_param->volume_format)) {
                setattr.lsv_clone.set_it = 1;
                setattr.lsv_clone.source = old_fileid->id;
                strcpy(setattr.lsv_clone.snap, clone_param->snap_name);
        } else {
                setattr.clone.set_it = 1;
                setattr.clone.val = 1;
        }
#else
        setattr.clone.set_it = 1;
        setattr.clone.val = 1;
#endif

        setattr.mode.set_it = 1;
        setattr.mode.val = clone_param->stbuf.st_mode;
        setattr.uid.set_it = 1;
        setattr.uid.val = clone_param->stbuf.st_uid;
        setattr.gid.set_it = 1;
        setattr.gid.val = clone_param->stbuf.st_gid;
        setattr.atime.set_it = __SET_TO_CLIENT_TIME;
        setattr.atime.time.seconds = clone_param->stbuf.st_atime;
        setattr.mtime.set_it = __SET_TO_CLIENT_TIME;
        setattr.mtime.time.seconds = clone_param->stbuf.st_mtime;

        vol_param_init(&param);
        param.priority = clone_param->priority;
        param.setattr = &setattr;
        strcpy(param.name, name);
        strcpy(param.site, clone_param->site);
        param.volume_format = clone_param->volume_format;

        // create new volume
        retry = 0;
retry1:
        ret = stor_mkvol_with_area(&newfile, &parent, &param);
        if (unlikely(ret)) {
                if (ret == EEXIST) {
                        if (retry) {
                                //pass
                                ret = stor_lookup(dest_pool, &parent, param.name, &newfile);
                                if (ret)
                                        GOTO(err_ret, ret);
                        } else
                                GOTO(err_ret, ret);
                } else if (ret == EAGAIN) {
                        ret = _errno(ret);
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if (clone_param->volume_format == VOLUME_FORMAT_RAW) {

                retry = 0;
        retry2:
                ret = __utils_snapshot_clone_table(old_fileid, &newfile);
                if (unlikely(ret)) {
                        ret = _errno(ret);
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_rm, ret, retry2, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                        } else
                                GOTO(err_rm, ret);
                }

                // TODO 如果此处失败
                // set source
        retry3:
                snprintf(src, sizeof(src), CHKID_FORMAT"/"CHKID_FORMAT,
                                CHKID_ARG(&clone_param->vol_fileid), CHKID_ARG(old_fileid));
                ret = md_xattr_set(NULL, &newfile, LICH_SYSTEM_ATTR_SOURCE, src, strlen(src) + 1, 0);
                if (unlikely(ret)) {
                        ret = _errno(ret);
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_rm, ret, retry3, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                        } else
                                GOTO(err_rm, ret);
                }
        } else {
                // LSV bitmap
        }

        return 0;
err_rm:
        UNIMPLEMENTED(__DUMP__);
err_ret:
        return ret;
}


static int __snapshot_clone(const char *src_pool, const char *dest_pool, snap_clone_param_t *clone_param)
{
        int ret;
        // fileid_t fileid, snapid;
        // struct stat stbuf;
        // fileinfo_t fileinfo;

        ret = utils_snapshot_src(src_pool, clone_param->vol_name,
                        clone_param->snap_name,
                        &clone_param->vol_fileid,
                        &clone_param->snap_fileid,
                        &clone_param->stbuf,
                        clone_param->volume_format);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (clone_param->volume_format == VOLUME_FORMAT_RAW) {
                ret = md_getattr(&clone_param->snap_fileid, &clone_param->snap_fileinfo);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }
        }

        ret = __utils_snapshot_clone(dest_pool, clone_param);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return abs(ret);
}

static int __snapshot_clone2(const char *src_pool, const char *dest_pool, snap_clone_param_t *clone_param)
{
        int ret;

        /**if (utils_spare_policy(src_pool)) {
          DERROR("reach the spare policy\n");
          ret = ENOSPC;
          GOTO(err_ret, ret);
          }**/

        // TODO LSV snap clone
        ret = stor_lookup1(src_pool, clone_param->vol_name, &clone_param->vol_fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = md_getattr(&clone_param->vol_fileid, &clone_param->vol_fileinfo);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = md_chunk_getsite(src_pool, &clone_param->vol_fileid, clone_param->site);
        if (ret)
                GOTO(err_ret, ret);

        clone_param->volume_format = lich_volume_format(&clone_param->vol_fileinfo);

        ret = __snapshot_clone(src_pool, dest_pool, clone_param);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

static int __group_snapshot_clone(const char *src_pool, const char *vol, const char *_snap,
                const char *dest_pool, const char *to, int p)
{
        int ret, vol_count, to_count, i, failed;
        char vol_tmp[MAX_GROUP_NAME_LEN], to_tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *vol_list[MAX_GROUP_VOLUMES_NUM], *to_list[MAX_GROUP_VOLUMES_NUM];
        snap_clone_param_t clone_param;

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vol, vol_tmp, vol_list, &vol_count);
        if (ret)
                GOTO(err_ret, ret);

        ret = __group_snapshot_split(to, to_tmp, to_list, &to_count);
        if (ret)
                GOTO(err_ret, ret);

        if (vol_count != to_count) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        for (i = 0; i < vol_count; i++) {
                ret = __is_snapshot_exist(src_pool, vol_list[i], snap);
                if (ret != 1) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = ENOENT;
                        GOTO(err_ret, ret);
                }
        }

        failed = 0;
        for (i = 0; i < vol_count; i++) {
                snap_clone_param_init(&clone_param);
                strcpy(clone_param.vol_name, vol_list[i]);
                strcpy(clone_param.snap_name, snap);
                strcpy(clone_param.to_vol_name, to_list[i]);
                clone_param.priority = p;
                ret = __snapshot_clone2(src_pool, dest_pool, &clone_param);
                if (ret) {
                        printf("failed clone %s@%s to %s, ret (%d)\n", vol_list[i], snap, to_list[i], ret);
                        failed = i;
                        GOTO(err_remove, ret);
                }
        }

        return 0;
err_remove:
        for (i = 0; i < failed; i++) {
                utils_rmvol(dest_pool, to_list[i], 0);
        }
err_ret:
        return ret;
}

int utils_snapshot_clone(const char *src_pool, const char *name, const char *dest_pool, const char *to, int p)
{
        int ret, count;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];
        snap_clone_param_t clone_param;

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                snap_clone_param_init(&clone_param);
                strcpy(clone_param.vol_name, list[0]);
                strcpy(clone_param.snap_name, list[1]);
                strcpy(clone_param.to_vol_name, to);
                clone_param.priority = p;

                ret = __snapshot_clone2(src_pool, dest_pool, &clone_param);
        } else {
                ret = __group_snapshot_clone(src_pool, list[0], list[1], dest_pool, to, p);
        }
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

static int __snapshot_protect(const char *pool, const char *vol, const char *snap)
{
        int ret;
        chkid_t chkid;
        snap_protect_param_t on;

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        snap_protect_param_init(&on);
        strcpy(on.snap_name, snap);
        on.on = 1;

        ret = stor_snapshot_protect(pool, &chkid, snap, on);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return (ret >= 0 ? ret : -ret);
}

static int __group_snapshot_protect(const char *pool, const char *vols, const char *_snap)
{
        int ret, count, i, failed = 0;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __is_snapshot_exist(pool, list[i], snap);
                if (ret != 1) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = ENOENT;
                        GOTO(err_ret, ret);
                }
        }

        for (i = 0; i < count; i++) {
                ret = __snapshot_protect(pool, list[i], snap);
                if (ret)
                        failed++;
        }

        if (failed) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_protect(const char *pool, const char *name)
{
        int ret, count;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                ret = __snapshot_protect(pool, list[0], list[1]);
        } else {
                ret = __group_snapshot_protect(pool, list[0], list[1]);
        }
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

static int __snapshot_unprotect(const char *pool, const char *vol, const char *snap)
{
        int ret;
        chkid_t chkid;
        snap_protect_param_t on;

        ret = stor_lookup1(pool, vol, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        snap_protect_param_init(&on);
        memcpy(on.snap_name,snap,strlen(snap));
        on.on = 0;

        ret = stor_snapshot_protect(pool, &chkid, snap, on);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return abs(ret);
}

static int __group_snapshot_unprotect(const char *pool, const char *vols, const char *_snap)
{
        int ret, count, i, failed = 0;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];

        GROUP(snap, _snap);

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __is_snapshot_exist(pool, list[i], snap);
                if ( ret != 1) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = ENOENT;
                        GOTO(err_ret, ret);
                }
        }

        for (i = 0; i < count; i++) {
                ret = __snapshot_unprotect(pool, list[i], snap);
                if (ret)
                        failed++;
        }

        if (failed) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_unprotect(const char *pool, const char *name)
{
        int ret, count;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                ret = __snapshot_unprotect(pool, list[0], list[1]);
        } else {
                ret = __group_snapshot_unprotect(pool, list[0], list[1]);
        }
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_protect_set(const char *pool, const char *name, int num)
{
        if (num)
                return utils_snapshot_protect(pool, name);
        else
                return utils_snapshot_unprotect(pool, name);
}

static int __snapshot_protect_get(const char *pool, const char *vol, const char *snap, int *protect)
{
        int ret;
        fileid_t fileid, chkid;
        fileinfo_t fileinfo;

        ret = stor_lookup1(pool, vol, &fileid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &fileid, snap, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = md_getattr(&chkid, &fileinfo);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if ((fileinfo.attr & __FILE_ATTR_PROTECT__) ==  __FILE_ATTR_PROTECT__) {
                *protect = 1;
        } else {
                *protect = 0;
        }

        return 0;
err_ret:
        return ret;
}

static int __group_snapshot_protect_get(const char *pool, const char *vols, const char *_snap, int *protect)
{
        int ret, count, i;
        char tmp[MAX_GROUP_NAME_LEN], snap[MAX_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];

        GROUP(snap, _snap);
        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __is_snapshot_exist(pool, list[i], snap);
                if ( ret != 1) {
                        if (ret < 0)
                                ret = -ret;
                        else
                                ret = ENOENT;
                        GOTO(err_ret, ret);
                }

                ret = __snapshot_protect_get(pool, list[i], snap, protect);
                if (ret)
                        GOTO(err_ret, ret);

                if (!*protect)
                        break;
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_protect_get(const char *pool, const char *name)
{
        int ret, count, protect;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[2];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        if (strchr(list[0], ',') == NULL) {
                ret = __snapshot_protect_get(pool, list[0], list[1], &protect);
        } else {
                ret = __group_snapshot_protect_get(pool, list[0], list[1], &protect);
        }
        if (ret)
                GOTO(err_ret, ret);

        printf("path: %s, protect: %d\n", name, protect);

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_cat(const char *pool, const char *name)
{
        int ret, count;
        uint64_t left, offset, size;
        char tmp[MAX_NAME_LEN], buf[BIG_BUF_LEN + 1];
        char *list[2];
        fileid_t parent, fileid;
        struct stat stbuf;
        lichbd_ioctx_t ioctx;

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = utils_snapshot_src(pool, list[0], list[1], &parent, &fileid, &stbuf, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = lichbd_connect(pool, list[0], &ioctx, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        offset = 0;
        left = stbuf.st_size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                ret = lichbd_snap_pread(&ioctx, &fileid, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        GOTO(err_ret, ret);
                }

                buf[size] = '\0';
                printf("%s", buf);

                offset += size;
                left -= size;
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_copy_single(lichbd_ioctx_t *ioctx, const fileid_t *snapid,
                struct stat *stbuf, const char *to)
{
        int ret, fd;
        char buf[BIG_BUF_LEN];
        uint64_t left, offset, size;


        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_ret, ret);
        }

        offset = 0;
        left = stbuf->st_size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                ret = lichbd_snap_pread(ioctx, snapid, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        GOTO(err_fd, ret);
                }

                ret = pwrite(fd, buf, size, offset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_fd, ret);
                }

                offset += size;
                left -= size;
        }

        close(fd);

        return 0;
err_fd:
        close(fd);
err_ret:
        return ret;
}

int utils_snapshot_copy_range(lichbd_ioctx_t *ioctx, const fileid_t *snapid,
                off_t _offset, size_t _size, const char *to)
{
        int ret, fd, retry, chknum, idx = 0, count = 0;
        char buf[BIG_BUF_LEN], *tmp;
        uint64_t left, offset, size, foffset;

        chknum = size2chknum(_size, NULL);
        ret = ymalloc((void **)&tmp, LICH_SNAPDIFF_HEAD_LEN(chknum));
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        memset(tmp, 0x0, LICH_SNAPDIFF_HEAD_LEN(chknum));
        memcpy(tmp, LICH_SNAPDIFF_MAGIC, strlen(LICH_SNAPDIFF_MAGIC));
        *(off_t *)(tmp + LICH_SNAPDIFF_OFFSET_OFF) = _offset;
        *(size_t *)(tmp + LICH_SNAPDIFF_SIZE_OFF) = _size;

        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_free, ret);
        }

        offset = _offset;
        left = _size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                retry = 0;
retry:
                ret = lichbd_snap_pread(ioctx, snapid, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_fd, ret, retry, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_fd, ret);
                }

                idx = (offset - _offset) / LICH_CHUNK_SPLIT;
                if (*(char *)(tmp + LICH_SNAPDIFF_BITMAP_OFF + idx) == 0) {
                        *(char *)(tmp + LICH_SNAPDIFF_BITMAP_OFF + idx) = 1;
                        count++;
                }

                foffset = LICH_SNAPDIFF_HEAD_LEN(chknum) +
                        (count - 1) * LICH_CHUNK_SPLIT +
                        offset % LICH_CHUNK_SPLIT;
                ret = pwrite(fd, buf, size, foffset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_fd, ret);
                }

                offset += size;
                left -= size;
        }

        ret = pwrite(fd, tmp, LICH_SNAPDIFF_HEAD_LEN(chknum), 0);
        if (ret < 0) {
                ret = errno;
                GOTO(err_fd, ret);
        }

        close(fd);
        yfree((void **)&tmp);

        return 0;
err_fd:
        close(fd);
err_free:
        yfree((void **)&tmp);
err_ret:
        return ret;
}

int utils_snapshot_copy_offset(lichbd_ioctx_t *ioctx, const fileid_t *snapid,
                off_t _offset, size_t _size, const char *to)
{
        int ret, fd, retry;
        char buf[BIG_BUF_LEN];
        uint64_t left, offset, size;

        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_ret, ret);
        }

        offset = _offset;
        left = _size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                retry = 0;
retry:
                ret = lichbd_snap_pread(ioctx, snapid, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_fd, ret, retry, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_fd, ret);
                }

                ret = pwrite(fd, buf, size, offset - _offset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_fd, ret);
                }

                offset += size;
                left -= size;
        }

        close(fd);

        return 0;
err_fd:
        close(fd);
err_ret:
        return ret;
}

void * __snapshot_copy_worker(void *_arg)
{
        int ret, retry;
        arg_t *arg;
        char buf[BIG_BUF_LEN];
        uint64_t left, offset, size;

        arg = _arg;

        offset = arg->offset;
        left = arg->size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                retry = 0;
retry:
                ret = lichbd_snap_pread(&arg->ext.ioctx, &arg->ext.snapid, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_ret, ret, retry, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_ret, ret);
                }

                ret = pwrite(arg->ext.fd, buf, size, offset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_ret, ret);
                }

                offset += size;
                left -= size;
        }

        arg->ret = 0;
        sem_post(&arg->sem);

        return NULL;
err_ret:
        arg->ret = ret;
        sem_post(&arg->sem);
        return NULL;
}

int utils_snapshot_copy_multi(lichbd_ioctx_t *ioctx, const fileid_t *snapid,
                struct stat *stbuf, const char *to)
{
        int ret, fd;
        arg_ext_t arg_ext;

        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_ret, ret);
        }

        arg_ext.fd = fd;
        arg_ext.ioctx = *ioctx;
        arg_ext.snapid = *snapid;
        arg_ext.worker = __snapshot_copy_worker;

        ret = __snapshot_copy_thread_startup(stbuf->st_size, &arg_ext);
        if (unlikely(ret))
                GOTO(err_fd, ret);


        close(fd);

        return 0;
err_fd:
        close(fd);
err_ret:
        return ret;
}

int utils_snapshot_copy(const char *pool, const char *name, const char *to, const char *idx)
{
        int ret, count, retry = 0, ifrom, icount;
        off_t offset;
        size_t size;
        char tmp[MAX_NAME_LEN], *p;
        char *list[2];
        fileid_t fileid, snapid;
        struct stat stbuf;
        lichbd_ioctx_t ioctx;

        /*
           if (utils_spare_policy(pool)) {
           DERROR("reach the spare policy\n");
           ret = ENOSPC;
           GOTO(err_ret, ret);
           }
           */

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

retry0:
        ret = utils_snapshot_src(pool, list[0], list[1], &fileid, &snapid, &stbuf, 0);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

retry1:
        ret = lichbd_connect(pool, list[0], &ioctx, 0);
        if (ret) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if (name[0] != ':' && to[0] == ':') {
                if (idx == NULL && MULTITHREADING) {
                        ret = utils_snapshot_copy_multi(&ioctx, &snapid, &stbuf, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                } else if (idx == NULL) {
                        ret = utils_snapshot_copy_single(&ioctx, &snapid, &stbuf, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                } else if (strchr(idx, '~')){
                        strcpy(tmp, idx);
                        p = strchr(tmp, '~');
                        *p = '\0';

                        ifrom = atoi(tmp);
                        icount = atoi(p + 1);

                        if (icount == 0) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        offset = ifrom * LICH_CHUNK_SPLIT;
                        if (offset >= stbuf.st_size) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        size = icount * LICH_CHUNK_SPLIT;
                        size = _min(size, (size_t)stbuf.st_size - offset);

                        ret = utils_snapshot_copy_range(&ioctx, &snapid, offset, size, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                } else {
                        offset = atoi(idx) * LICH_CHUNK_SPLIT;
                        if (offset >= stbuf.st_size) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        size = stbuf.st_size - offset;
                        size = size > LICH_CHUNK_SPLIT ? LICH_CHUNK_SPLIT : size;

                        ret = utils_snapshot_copy_offset(&ioctx, &snapid, offset, size, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                }
        } else {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_diff_range(lichbd_ioctx_t *ioctx, const fileid_t *snapsrc,
                const fileid_t *snapdst, off_t _offset, size_t _size, const char *to)
{
        int ret, fd, retry, chknum, idx = 0, count = 0;
        char buf[BIG_BUF_LEN], *tmp;
        uint64_t left, offset, size, foffset;

        chknum = size2chknum(_size, NULL);
        ret = ymalloc((void **)&tmp, LICH_SNAPDIFF_HEAD_LEN(chknum));
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        memset(tmp, 0x0, LICH_SNAPDIFF_HEAD_LEN(chknum));
        memcpy(tmp, LICH_SNAPDIFF_MAGIC, strlen(LICH_SNAPDIFF_MAGIC));
        *(off_t *)(tmp + LICH_SNAPDIFF_OFFSET_OFF) = _offset;
        *(size_t *)(tmp + LICH_SNAPDIFF_SIZE_OFF) = _size;

        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_free, ret);
        }

        offset = _offset;
        left = _size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                retry = 0;
retry:
                ret = lichbd_snap_diff(ioctx, snapsrc, snapdst, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        if (ret == ENOENT) {
                                offset += size;
                                left -= size;
                                continue;
                        } if (ret == EAGAIN) {
                                USLEEP_RETRY(err_fd, ret, retry, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_fd, ret);
                }

                idx = (offset - _offset) / LICH_CHUNK_SPLIT;
                if (*(char *)(tmp + LICH_SNAPDIFF_BITMAP_OFF + idx) == 0) {
                        *(char *)(tmp + LICH_SNAPDIFF_BITMAP_OFF + idx) = 1;
                        count++;
                }

                foffset = LICH_SNAPDIFF_HEAD_LEN(chknum) +
                        (count - 1) * LICH_CHUNK_SPLIT +
                        offset % LICH_CHUNK_SPLIT;
                ret = pwrite(fd, buf, size, foffset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_fd, ret);
                }

                offset += size;
                left -= size;
        }

        ret = pwrite(fd, tmp, LICH_SNAPDIFF_HEAD_LEN(chknum), 0);
        if (ret < 0) {
                ret = errno;
                GOTO(err_fd, ret);
        }

        close(fd);
        yfree((void **)&tmp);

        return 0;
err_fd:
        close(fd);
err_free:
        yfree((void **)&tmp);
err_ret:
        return ret;
}

int utils_snapshot_diff_offset(lichbd_ioctx_t *ioctx, const fileid_t *snapsrc,
                const fileid_t *snapdst, off_t _offset, size_t _size, const char *to)
{
        int ret, fd, retry;
        char buf[BIG_BUF_LEN];
        uint64_t left, offset, size;

        fd = open(to, O_WRONLY | O_CREAT | O_TRUNC, 0644);
        if (fd < 0) {
                ret = errno;
                GOTO(err_ret, ret);
        }

        offset = _offset;
        left = _size;
        while (left > 0) {
                size = left < BIG_BUF_LEN ? left : BIG_BUF_LEN;

                retry = 0;
retry:
                ret = lichbd_snap_diff(ioctx, snapsrc, snapdst, buf, size, offset);
                if (ret < 0) {
                        ret = -ret;
                        if (ret == EAGAIN) {
                                USLEEP_RETRY(err_fd, ret, retry, retry, 50, (100 * 1000));
                        } else
                                GOTO(err_fd, ret);
                }

                ret = pwrite(fd, buf, size, offset - _offset);
                if (ret < 0) {
                        ret = errno;
                        GOTO(err_fd, ret);
                }

                offset += size;
                left -= size;
        }

        close(fd);

        return 0;
err_fd:
        close(fd);
err_ret:
        return ret;
}

static int __utils_snapshot_diff_split(const char *pool, const char *name, char *file,
                fileid_t *fileid, fileid_t *snapsrc, fileid_t *snapdst, char *namesrc, char *namedst,
                uint64_t *ver_src, uint64_t *ver_dst, struct stat *stbuf)
{
        int ret, count, retry = 0;
        char tmp[MAX_NAME_LEN];
        char *list[3];

        count = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        count = 2;
        _str_split(list[1], '~', &list[1], &count);
        if (count != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        strcpy(file, list[0]);
        if (snapsrc)
                strcpy(namesrc, list[1]);
        if (snapdst)
                strcpy(namedst, list[2]);

retry0:
        ret = utils_snapshot_src(pool, file, namesrc, fileid, snapsrc, stbuf, 0);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        ret = stor_getsnapversion(pool, snapsrc, ver_src);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

retry1:
        ret = utils_snapshot_src(pool, file, namedst, fileid, snapdst, stbuf, 0);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        ret = stor_getsnapversion(pool, snapdst, ver_dst);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int __snapshot_flat(const char *pool, const char *name, int force)
{
        int ret;
        chkid_t fileid;

        ret = stor_lookup1(pool, name, &fileid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_snapshot_flat(&fileid, 1, force);
        if (ret) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __group_snapshot_flat(const char *pool, const char *vols, int force)
{
        int ret, count, i, failed = 0;
        char tmp[MAX_GROUP_NAME_LEN];
        char *list[MAX_GROUP_VOLUMES_NUM];


        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        for (i = 0; i < count; i++) {
                ret = __snapshot_flat(pool, list[i], force);
                if (ret) {
                        failed++;
                        printf("flat %s failed, %s\n", list[i], strerror(ret));
                } else {
                        printf("flat %s success\n", list[i]);
                }
        }

        if (failed) {
                ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int __utils_snapshot_flat(const char *pool, const char *name, int force)
{
        int ret;

        if (strchr(name, ',') == NULL)
                ret = __snapshot_flat(pool, name, force);
        else
                ret = __group_snapshot_flat(pool, name, force);
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_flat(const char *pool, const char *name)
{
        int ret, force, retry;

        force = 0;
        retry = 0;
retry:
        ret = __utils_snapshot_flat(pool, name, force);
        if (ret) {
                ret = _errno(ret);
                if (ret == EAGAIN) {
                        force = 1;
                        USLEEP_RETRY(err_ret, ret, retry, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                } else if (ret == EEXIST) {
                        YASSERT(force == 1 && retry > 0);
                        //no op;
                } else
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff(const char *pool, const char *name, const char *to, const char *idx)
{
        int ret, retry = 0, ifrom, icount;
        char tmp[MAX_NAME_LEN], *p;
        off_t offset;
        size_t size;
        fileid_t fileid, snapdst, snapsrc;
        uint64_t ver_src, ver_dst;
        struct stat stbuf;
        lichbd_ioctx_t ioctx;

        ret = __utils_snapshot_diff_split(pool, name, tmp, &fileid, &snapsrc, &snapdst,
                        NULL, NULL, &ver_src, &ver_dst, &stbuf);
        if (ret)
                GOTO(err_ret, ret);

        if (ver_src >= ver_dst) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

retry2:
        ret = lichbd_connect(pool, tmp, &ioctx, 0);
        if (ret) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry2, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if (name[0] != ':' && to[0] == ':') {
                if (idx == NULL) {
                        /* head max size: LICH_CHUNK_SPLIT */
                        if (size2chknum(stbuf.st_size, NULL) > LICH_CHUNK_SPLIT) {
                                ret = EFBIG;
                                GOTO(err_ret, ret);
                        }

                        /* out put with head & data */
                        ret = utils_snapshot_diff_range(&ioctx, &snapsrc, &snapdst, 0, stbuf.st_size, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                } else if (strchr(idx, '~')){
                        strcpy(tmp, idx);
                        p = strchr(tmp, '~');
                        *p = '\0';

                        ifrom = atoi(tmp);
                        icount = atoi(p + 1);

                        if (icount == 0) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        offset = ifrom * LICH_CHUNK_SPLIT;
                        if (offset >= stbuf.st_size) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        size = icount * LICH_CHUNK_SPLIT;
                        size = _min(size, (size_t)stbuf.st_size - offset);

                        /* out put with head & data */
                        ret = utils_snapshot_diff_range(&ioctx, &snapsrc, &snapdst, offset, size, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                } else {
                        offset = atoi(idx) * LICH_CHUNK_SPLIT;
                        if (offset >= stbuf.st_size) {
                                ret = EINVAL;
                                GOTO(err_ret, ret);
                        }

                        size = stbuf.st_size - offset;
                        size = size > LICH_CHUNK_SPLIT ? LICH_CHUNK_SPLIT : size;

                        /* out put not need head, only data direct */
                        ret = utils_snapshot_diff_offset(&ioctx, &snapsrc, &snapdst, offset, size, to + 1);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                }
        } else {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff_vol_range(const char *pool, const chkid_t *fileid,
                buffer_t *buf, const int start, const int count)
{
        int ret, i, begin, end;
        chkid_t tableid, chkid;
        uint64_t left, offset, size, boff, bsize;
        buffer_t tmp;

        (void) pool;

        ret = mbuffer_init(&tmp, 0);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        YASSERT(start % CHAR_BIT == 0);
        begin = start / FILE_PROTO_EXTERN_ITEM_COUNT;
        end = (start + count) / FILE_PROTO_EXTERN_ITEM_COUNT;

        offset = start;
        left = count;
        /* read bimap for loop in subvols */
        for (i = begin; i <= end; i++) {
                fid2cid(&chkid, fileid, offset);
                cid2tid(&tableid, &chkid);

                boff = (offset % FILE_PROTO_EXTERN_ITEM_COUNT) / CHAR_BIT;
                size = _min(FILE_PROTO_EXTERN_ITEM_COUNT - offset % FILE_PROTO_EXTERN_ITEM_COUNT, left);
                bsize = _ceil(size, CHAR_BIT);

                ret = md_chunk_table_read(fileid, &tableid, &tmp, bsize, TABLE_PROTO_MAP + boff);
                if (unlikely(ret)) {
                        if (ret == ENOENT)
                                mbuffer_appendzero(&tmp, bsize);
                        else
                                GOTO(err_free, ret);
                }

                mbuffer_merge(buf, &tmp);

                left -= size;
                offset += size;
        }

        YASSERT(buf->len == _ceil(count, CHAR_BIT));

        return 0;
err_free:
        mbuffer_free(&tmp);
err_ret:
        return 0;
}

STATIC int __utils_snapshot_diff_vol(const char *pool, const char *name, const char *idx,
                buffer_t *buf, int *total, int *start, int *count)
{
        int ret, retry = 0, ifrom, icount;
        fileid_t fileid;
        struct stat stbuf;
        char tmp[MAX_NAME_LEN], *p;
        lichbd_ioctx_t ioctx;

        ret = stor_lookup1(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = stor_getattr(pool, &fileid, &stbuf);
        if (unlikely(ret))
                GOTO(err_ret, ret);

retry1:
        ret = lichbd_connect(pool, name, &ioctx, 0);
        if (ret) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry1, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if (idx == NULL) {
                ifrom = 0;
                icount = size2chknum(stbuf.st_size, NULL);
        } else if (strchr(idx, '~')){
                strcpy(tmp, idx);
                p = strchr(tmp, '~');
                *p = '\0';

                ifrom = atoi(tmp);
                icount = atoi(p + 1);

                if (icount == 0) {
                        ret = EINVAL;
                        GOTO(err_ret, ret);
                }
        } else {
                ifrom = atoi(tmp);
                icount = 1;
        }

        ret = __utils_snapshot_diff_vol_range(pool, &fileid, buf, ifrom, icount);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        *total = size2chknum(stbuf.st_size, NULL);
        *start = ifrom;
        *count = icount;

        return 0;
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff_volsnap(const char *pool, const char *name, const char *idx,
                buffer_t *buf, int *total, int *start, int *count)
{
        int ret, lcount, retry = 0;
        char tmp[MAX_NAME_LEN];
        char *list[2];
        struct stat stbuf;
        fileid_t fileid, snapsrc, snaplast;
        nid_t snapnid;
        char snapname[MAX_NAME_LEN];
        uint64_t ver_src, ver_last;

        lcount = 2;
        strcpy(tmp, name);
        _str_split(tmp, '@', list, &lcount);
        if (lcount != 2) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

retry0:
        ret = utils_snapshot_src(pool, list[0], list[1], &fileid, &snapsrc, &stbuf, 0);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        ret = stor_getsnapversion(pool, &snapsrc, &ver_src);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry0, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        ret = md_snapshot_last(&fileid, &snapnid, &snaplast, snapname, &ver_last);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        /* only allow different between the volume and the last snapshot of volume */
        if (ver_last != ver_src) {
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        tmp[strlen(list[0])] = '/';
        ret = __utils_snapshot_diff_vol(pool, tmp, idx, buf, total, start, count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff_2snaps(const char *pool, const char *name, const char *idx,
                buffer_t *buf, int *total, int *start, int *count)
{
        int ret;
        char tmp[MAX_NAME_LEN];

        fileid_t fileid, snapdst, snapsrc, snapprev;
        char namesrc[MAX_NAME_LEN], namedst[MAX_NAME_LEN], nameprev[MAX_NAME_LEN];
        uint64_t ver_src, ver_dst, ver_prev;
        struct stat stbuf;

        ret = __utils_snapshot_diff_split(pool, name, tmp, &fileid, &snapsrc, &snapdst,
                        namesrc, namedst, &ver_src, &ver_dst, &stbuf);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = md_snapshot_prev(&fileid, &snapdst, &snapprev, nameprev, &ver_prev);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        /* the 2snaps must is parent and son */
        if (ver_src != ver_prev) {
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        strcat(tmp, "/");
        strcat(tmp, namesrc);
        ret = __utils_snapshot_diff_vol(pool, tmp, idx, buf, total, start, count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff_base64encode(const buffer_t *buf, char **bitmap)
{
        int ret;
        char *tmp, *code;

        ret = ymalloc((void **)&tmp, buf->len);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = ymalloc((void **)&code, buf->len * 2);
        if (unlikely(ret))
                GOTO(err_free, ret);

        mbuffer_get(buf, tmp, buf->len);
        base64_encode((void *)tmp, buf->len, code);

        *bitmap = code;
        yfree((void **)&tmp);

        return 0;
err_free:
        yfree((void **)&tmp);
err_ret:
        return ret;
}

STATIC int __utils_snapshot_diff_output(int total, int start, int count, const char *bitmap)
{
        cJSON *json = cJSON_CreateObject();

        /* {"blksize":1048576, "blocks":xxxxxx, "bitmap":"base64code", "start":xxx, "count":xxx, "used":xxx} */

        cJSON_AddNumberToObject(json, "blksize", 1048576);
        cJSON_AddNumberToObject(json, "blocks", total);
        cJSON_AddNumberToObject(json, "start", start);
        cJSON_AddNumberToObject(json, "count", count);
        cJSON_AddStringToObject(json, "bitmap", bitmap);

        printf("%s\n", cJSON_PrintUnformatted(json));

        //释放json结构所占用的内存
        cJSON_Delete(json);
        return 0;
}

/**
 * get bit map output with json, format:
 * {"blksize":1048576, "blocks":xxxxxx, "bitmap":"base64code", "start":xxx, "count":xxx, "used":xxx}
 */
STATIC int __utils_snapshot_diff_json(const char *pool, const char *name, const char *to, const char *idx)
{
        int ret, total, start, count;
        buffer_t buf;
        char *bitmap;

        (void) to;

        ret = mbuffer_init(&buf, 0);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (strchr(name, '@') && strchr(name, '~')) {
                ret = __utils_snapshot_diff_2snaps(pool, name, idx, &buf, &total, &start, &count);
                if (unlikely(ret))
                        GOTO(err_free, ret);
        } else if (strchr(name, '@')) {
                ret = __utils_snapshot_diff_volsnap(pool, name, idx, &buf, &total, &start, &count);
                if (unlikely(ret))
                        GOTO(err_free, ret);
        } else {
                ret = __utils_snapshot_diff_vol(pool, name, idx, &buf, &total, &start, &count);
                if (unlikely(ret))
                        GOTO(err_free, ret);
        }

        ret = __utils_snapshot_diff_base64encode(&buf, &bitmap);
        if (unlikely(ret))
                GOTO(err_free, ret);

        ret = __utils_snapshot_diff_output(total, start, count, bitmap);
        if (unlikely(ret))
                GOTO(err_free, ret);

        mbuffer_free(&buf);
        yfree((void **)&bitmap);

        return 0;
err_free:
        mbuffer_free(&buf);
err_ret:
        return ret;
}

int utils_snapshot_diff(const char *pool, const char *name, const char *to, const char *idx, const int json)
{
        if (json) /* only get bitmap output with json */
                return __utils_snapshot_diff_json(pool, name, to, idx);
        else /* get bitmap and data download to 'to' */
                return __utils_snapshot_diff(pool, name, to, idx);
}

#define MAX_LEVEL 1024

typedef struct {
        struct list_head hook;
        char snap_name[MAX_NAME_LEN];
        char parent_name[MAX_NAME_LEN];
        long snap_ver;
        long snap_from;
        BOOL deleting;
        int is_last;
        struct list_head child;
} list_node_t;

static struct list_head tree_list;
static list_node_t tree_root;
static fileid_t tree_id;
static char dirs[MAX_LEVEL];

static void __snapshot_list_init(const char *name)
{
        strcpy(tree_root.snap_name, name);
        tree_root.snap_ver = -1;
        tree_root.deleting = 0;
        INIT_LIST_HEAD(&tree_root.child);
        INIT_LIST_HEAD(&tree_list);
}

static int __snapshot_list_insert(const char *snap_name, long snap_ver,
		const char *parent_name, long snap_from, BOOL deleting)
{
        int ret;
        list_node_t *node;

        ret = ymalloc((void **)&node, sizeof(list_node_t));
        if (unlikely(ret))
                GOTO(err_ret, ret);

        strcpy(node->snap_name, snap_name);
        strcpy(node->parent_name, parent_name);
        node->snap_ver = snap_ver;
        node->snap_from = snap_from;
        node->deleting = deleting;
        node->is_last = 1;
        INIT_LIST_HEAD(&node->child);

        list_add_tail(&node->hook, &tree_list);

        return 0;
err_ret:
        return ret;
}

static void __snapshot_list_redirect(long from, long to)
{
        list_node_t *node;
        struct list_head *pos, *n;

        list_for_each_safe(pos, n, &tree_list) {
                node = (void *)pos;
                if (node->snap_from == from)
                        node->snap_from = to;
        }
}

static void __snapshot_list_build(list_node_t *curr)
{
        list_node_t *node, *prev;
        struct list_head *pos, *n;

        list_for_each_safe(pos, n, &tree_list) {
                node = (void *)pos;
                if (node->snap_from == curr->snap_ver) {
                        list_del(pos);
                        if (node->deleting) {
                                __snapshot_list_redirect(node->snap_ver, curr->snap_ver);
                        } else {

                                list_add_tail(&node->hook, &curr->child);
                                prev = (list_node_t *)node->hook.prev;
                                if (prev != (list_node_t *)&curr->child)
                                        prev->is_last = 0;
                        }
                }
        }

        if (list_empty(&tree_list))
                return;

        list_for_each(pos, &curr->child) {
                node = (void *)pos;
                __snapshot_list_build(node);
        }
}

static int __snapshot_list_show_allocate(const char *pool, const char *name, uint64_t *allocate, uint64_t *rollback)
{
        int ret, multi = 1, full = 1;
        fileid_t fileid;
        chkinfo_t *chkinfo;
        char _chkinfo[CHKINFO_MAX];
        filestat_t filestat;

        if (strncmp(name, LICH_SYSTEM_ATTR_CURR, strlen(LICH_SYSTEM_ATTR_CURR))) {
                ret = stor_lookup(pool, &tree_id, name, &fileid);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }
        } else {
                fileid = tree_id;
        }

        chkinfo = (void *)_chkinfo;
        ret = md_chunk_getinfo(pool, NULL, &fileid, chkinfo, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = stor_stat_multi(CHKINFO_FIRSTNID(chkinfo), &fileid, &filestat, multi, full);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        *allocate = filestat.chknum - filestat.sparse;
        *rollback = filestat.rollback;

        return 0;
err_ret:
        return ret;
}

static void __snapshot_list_show_root(list_node_t *curr, int *show)
{
        list_node_t *node;
        struct list_head *pos;

        if (!strstr(curr->snap_name, LICH_SYSTEM_ATTR_ROOT) && curr != &tree_root) {
                *show = 1;
        } else {
                list_for_each(pos, &curr->child) {
                        node = (void *)pos;
                        __snapshot_list_show_root(node, show);
                }
        }
}

static void __snapshot_list_show_tree(const char *pool, list_node_t *curr, int level, int verbose, int all)
{
        int ret, i, show = 0;
        list_node_t *node;
        struct list_head *pos, *n;
        uint64_t allocate, rollback;

        if (level != 0) {
                if (curr->is_last)
                        dirs[level - 1] = 2;
                else
                        dirs[level - 1] = 1;

                memset(dirs + level, 0x0, MAX_LEVEL - level);
        }

        if (curr == &tree_root) {
                __snapshot_list_show_root(curr, &show);
                if (show)
                        printf("%s\n", curr->snap_name);
        } else {
                if (!all) {
                        if (strstr(curr->snap_name, LICH_SYSTEM_ATTR_ROOT)) {
                                list_for_each_safe(pos, n, &curr->child) {
                                        node = (void *)pos;
                                        __snapshot_list_show_tree(pool, node, level, verbose, all);

                                        list_del(pos);
                                        yfree((void **)&pos);
                                }

                                return;
                        }
                }

                for(i = 0; dirs[i] && i <= MAX_LEVEL; i++) {
                        printf("%s ", dirs[i + 1] ? (dirs[i]==1 ? "|  " : "   " ) :
                                        (dirs[i]==1 ? "|--" : "`--"));
                }

                if (verbose) {
                        ret = __snapshot_list_show_allocate(pool, curr->snap_name, &allocate, &rollback);
                        if (ret) {
                                printf("%s ver:%ld from:%ld allocate:N/A\n",
                                                curr->snap_name, curr->snap_ver, curr->snap_from);
                        } else {
                                if (strncmp(curr->snap_name, LICH_SYSTEM_ATTR_CURR"->", strlen(LICH_SYSTEM_ATTR_CURR"->"))) {
                                        printf("%s ver:%ld from:%ld allocate:%ld\n",
                                                        curr->snap_name, curr->snap_ver, curr->snap_from, allocate);
                                } else {
                                        printf("%s ver:%ld from:%ld allocate:%ld rollback:%ld\n",
                                                        curr->snap_name, curr->snap_ver, curr->snap_from,
                                                        allocate, rollback);
                                }
                        }
                } else
                        printf("%s\n", curr->snap_name);
        }

        list_for_each_safe(pos, n, &curr->child) {
                node = (void *)pos;
                __snapshot_list_show_tree(pool, node, level + 1, verbose, all);

                list_del(pos);
                yfree((void **)&pos);
        }
}

static void __snapshot_list_show_json(list_node_t *curr, int level, int verbose, int all)
{
        list_node_t *node;
        struct list_head *pos, *n;

        if (level != 0) {
                if (curr->is_last)
                        dirs[level - 1] = 2;
                else
                        dirs[level - 1] = 1;

                memset(dirs + level, 0x0, MAX_LEVEL - level);
        }

        if (curr == &tree_root) {
                printf("{\"filename\": \"%s\", \"snapshot\": {", curr->snap_name);
        } else {
                if (!all) {
                        if (strstr(curr->snap_name, LICH_SYSTEM_ATTR_ROOT)) {
                                list_for_each_safe(pos, n, &curr->child) {
                                        node = (void *)pos;
                                        __snapshot_list_show_json(node, level, verbose, all);

                                        list_del(pos);
                                        yfree((void **)&pos);
                                }

                                return;
                        }
                }

                if (verbose)
                        printf("\"%s\": {\"snapname\": \"%s\", \"ver\": %ld, \"from\": %ld ",
                                        curr->snap_name, curr->snap_name, curr->snap_ver, curr->snap_from);
                else
                        printf("\"%s\": {\"snapname\": \"%s\"", curr->snap_name, curr->snap_name);

                if (!list_empty(&curr->child)) {
                        printf(", \"child\": {");
                }
        }

        list_for_each_safe(pos, n, &curr->child) {
                node = (void *)pos;
                __snapshot_list_show_json(node, level + 1, verbose, all);
        }

        if (!list_empty(&curr->child)) {
                printf("}");
        }

        if (curr->is_last || curr == &tree_root)
                printf("}");
        else
                printf("}, ");

        list_for_each_safe(pos, n, &curr->child) {
                node = (void *)pos;

                list_del(pos);
                yfree((void **)&pos);
        }
}

static void __snapshot_list_show(const char *pool, int verbose, int all, int format)
{
        __snapshot_list_build(&tree_root);

        if (format == SNAPSHOT_SHOW_JSON)
                __snapshot_list_show_json(&tree_root, 0, verbose, all);
        else if (format == SNAPSHOT_SHOW_TREE)
                __snapshot_list_show_tree(pool, &tree_root, 0, verbose, all);
}

/**
 * format:  0 -- lich
 *          1 -- json
 *          2 -- tree
 */
static int __snapshot_list(const char *pool, const char *name, int all, int format, int verbose)
{
        int ret, done = 0;
        chkid_t chkid;

        char buf[BIG_BUF_LEN], *group;
        int buflen = BIG_BUF_LEN;
        struct dirent *de;
        //uint32_t  type_and_flag;
        uint64_t offset = 0, offset2 = 0;
        char uuid[MAX_NAME_LEN] = {};

        char snap_name[MAX_NAME_LEN];
        char parent_name[MAX_NAME_LEN] = {0};
        char rollback_name[MAX_NAME_LEN];
        long snap_ver, snap_from;
        fileinfo_t fileinfo;
	volume_format_t vft;

        ret = stor_lookup1(pool, name, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (chkid.type == __POOL_CHUNK__) {
                ret = EISDIR;
                GOTO(err_ret, ret);
        }

        tree_id = chkid;

        ret = md_getattr(&chkid, &fileinfo);
        if (ret)
                GOTO(err_ret, ret);

	vft = lich_volume_format(&fileinfo);
        __snapshot_list_init(name);
        get_uuid(uuid);

        ret = stor_snapshot_listopen(&chkid, uuid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        while (done == 0) {
                ret = stor_snapshot_list(&chkid, uuid, offset, buf, &buflen);
                if (unlikely(ret)) {
                        GOTO(err_close, ret);
                }

                if (buflen == 0)
                        break;

                offset2 = 0;
                dir_for_each(buf, buflen, de, offset2) {
                        if (strlen(de->d_name) == 0) {
                                done = 1;
                                goto out;
                        } else if (buflen - offset2 < sizeof(*de) + MAX_NAME_LEN) {
                                break;
                        }

                        offset += de->d_reclen;

                        if (strlen(GROUPFLAG)) {
                                group = strstr(de->d_name, GROUPFLAG);
                                if (!all && group && strlen(group) == strlen(de->d_name))
                                        continue;
                        }

                        if (IS_RICH_LSV(vft)) {
                                ret = sscanf(de->d_name, "%s %ld %s %ld", snap_name, &snap_ver, parent_name, &snap_from);
                                if (ret != 4)
                                        continue;
                        } else {
                                ret = sscanf(de->d_name, "%s %ld %ld", snap_name, &snap_ver, &snap_from);
                                if (ret != 3)
                                        continue;
                        }

                        DBUG("snap:%s version:%ld from:%ld ret:%d\n", snap_name, snap_ver, snap_from, ret);
                        if (!memcmp(snap_name, LICH_SYSTEM_ATTR_UNLINK, strlen(LICH_SYSTEM_ATTR_UNLINK))) {
                                if (all) {
                                        if (format) {
                                                ret = __snapshot_list_insert(snap_name, snap_ver, parent_name, snap_from, FALSE);
                                                if (unlikely(ret))
                                                        continue;
                                        } else {
                                                if (IS_RICH_LSV(vft))
                                                        printf("%s %s\n", snap_name, parent_name);
                                                else
                                                        printf("%s\n", snap_name);
                                        }
                                } else {

                                        ret = __snapshot_list_insert(snap_name, snap_ver, parent_name, snap_from, TRUE);
                                        if (unlikely(ret))
                                                continue;
                                }
                        } else if (!memcmp(snap_name, LICH_SYSTEM_ATTR_AUTO, strlen(LICH_SYSTEM_ATTR_AUTO))) {
                                if (all) {
                                        if (format) {
                                                ret = __snapshot_list_insert(snap_name, snap_ver, parent_name, snap_from, FALSE);
                                                if (unlikely(ret))
                                                        continue;
                                        } else {
                                                if (IS_RICH_LSV(vft))
                                                        printf("%s %s\n", snap_name, parent_name);
                                                else
                                                        printf("%s\n", snap_name);
                                        }
                                }
                        } else {
                                if (snap_ver == fileinfo.snap_rollback)
                                        strcpy(rollback_name, snap_name);

                                if (format) {
                                        ret = __snapshot_list_insert(snap_name, snap_ver, parent_name, snap_from, FALSE);
                                        if (unlikely(ret))
                                                continue;
                                } else {
                                        if (memcmp(snap_name, LICH_SYSTEM_ATTR_ROOT, strlen(LICH_SYSTEM_ATTR_ROOT))) {
                                                if (IS_RICH_LSV(vft))
                                                        printf("%s %s\n", snap_name, parent_name);
                                                else
                                                        printf("%s\n", snap_name);
                                        }
                                }
                        }
                }
        }

out:
        if (format) {
		if (!IS_RICH_LSV(format)) {
			if (fileinfo.snap_version != fileinfo.snap_rollback) {
				snprintf(snap_name, MAX_NAME_LEN, "%s->%s", LICH_SYSTEM_ATTR_CURR, rollback_name);
				__snapshot_list_insert(snap_name, fileinfo.snap_version, parent_name, fileinfo.snap_from, FALSE);
			} else {
				__snapshot_list_insert(LICH_SYSTEM_ATTR_CURR, fileinfo.snap_version, parent_name, fileinfo.snap_from, FALSE);
			}
		}
	}

        if (format)
                __snapshot_list_show(pool, verbose, all, format);

        stor_snapshot_listclose(&chkid, uuid);

        return 0;
err_close:
        stor_snapshot_listclose(&chkid, uuid);
err_ret:
        return ret;
}

static int __group_snapshot_list(const char *pool, const char *vols, int all, int format)
{
        int ret, count, i, found;
        char tmp[MAX_GROUP_NAME_LEN], *group;
        char *list[MAX_GROUP_VOLUMES_NUM];

        char buf[BIG_BUF_LEN];
        int buflen = BIG_BUF_LEN;
        struct dirent *de;
        uint64_t offset;

        char parent_name[MAX_NAME_LEN] = {0};
        char snap_name[MAX_NAME_LEN], rollback_name[MAX_NAME_LEN];
        long snap_ver, snap_from;
        fileinfo_t fileinfo;
	volume_format_t vft;

        (void)all;
        (void)format;

        ret = __group_snapshot_split(vols, tmp, list, &count);
        if (ret)
                GOTO(err_ret, ret);

        ret = __snapshot_list_one(pool, list[0], buf, &buflen, &fileinfo);
        if (ret)
                GOTO(err_ret, ret);

	vft = lich_volume_format(&fileinfo);
        __snapshot_list_init(vols);

        dir_for_each(buf, buflen, de, offset) {
                if (strlen(de->d_name) == 0) {
                        break;
                }

                if (!strcmp(de->d_name, ".") || !strcmp(de->d_name, "..")) {
                        continue;
                }

                if (IS_RICH_LSV(vft)) {
                        ret = sscanf(de->d_name, "%s %ld %s %ld", snap_name, &snap_ver, parent_name, &snap_from);
                        if (ret != 4)
                                continue;
                } else {
                        ret = sscanf(de->d_name, "%s %ld %ld", snap_name, &snap_ver, &snap_from);
                        if (ret != 3)
                                continue;
                }

                DBUG("snap:%s version:%ld from:%ld\n", snap_name, snap_ver, snap_from);
                if ((uint64_t)snap_ver == fileinfo.snap_rollback)
                        strcpy(rollback_name, snap_name);

                if (strncmp(snap_name, LICH_SYSTEM_ATTR_UNLINK, strlen(LICH_SYSTEM_ATTR_UNLINK))) {
                        found = 0;
                        for (i = 1; i < count; i++) {
                                ret = __is_snapshot_exist(pool, list[i], snap_name);
                                if (ret < 0) {
                                        ret = -ret;
                                        GOTO(err_ret, ret);
                                }

                                if (ret == 1) {
                                        found++;
                                }
                        }

                        DBUG("found:%d count:%d\n", found, count);
                        if (found != count -1)
                                continue;
                }

                group = strstr(snap_name, GROUPFLAG);
                if (group && strlen(group) == strlen(snap_name)) {
                        ret = __snapshot_list_insert(snap_name, snap_ver, parent_name, snap_from, FALSE);
                        if (unlikely(ret))
                                continue;
                }
        }

        if (all) {
		if (!IS_RICH_LSV(format)) {
                        if (fileinfo.snap_version != fileinfo.snap_rollback) {
                                snprintf(snap_name, MAX_NAME_LEN, "%s->%s", LICH_SYSTEM_ATTR_CURR, rollback_name);
                                __snapshot_list_insert(snap_name, fileinfo.snap_version, parent_name, fileinfo.snap_from, FALSE);
                        } else {
                                __snapshot_list_insert(LICH_SYSTEM_ATTR_CURR, fileinfo.snap_version, parent_name, fileinfo.snap_from, FALSE);
                        }
                }
        }

        __snapshot_list_show(pool, 0, all, format);

        return 0;
err_ret:
        return ret;
}

int utils_snapshot_list(const char *pool, const char *name, int all, int format, int verbose)
{
        int ret;

        if (strchr(name, ',') == NULL)
                ret = __snapshot_list(pool, name, all, format, verbose);
        else
                ret = __group_snapshot_list(pool, name, all, format);
        if (ret)
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

/**
 * @note: 检查单个快照是否处于被保护状态
 * @param
 *     path - /iscsi/v1@s1
 *     protect - 0, 未被保护;  1,被保护
 * @return
 *     正常返回0
 *     异常返回errno
 **/
int utils_snapshot_check_protect(const char *pool, const char *path, int *protect)
{
        int ret;
        fileid_t fileid;
        chkinfo_t *chkinfo;
        char _chkinfo[CHKINFO_MAX];
        filestat_t filestat;

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        chkinfo = (void *)_chkinfo;
        ret = md_chunk_getinfo(pool, NULL, &fileid, chkinfo, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = stor_stat_multi(CHKINFO_FIRSTNID(chkinfo), &fileid, &filestat, 1, 0);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (filestat.attr & __FILE_ATTR_PROTECT__)
                *protect = 1;
        else
                *protect = 0;

        return 0;
err_ret:
        return ret;
}

int utils_volume_has_protected_snapshot(const char *pool, const char *name)
{
        int ret, done = 0, protect = 0;
        chkid_t chkid;

        char buf[BIG_BUF_LEN];
        int buflen = BIG_BUF_LEN;
        struct dirent *de;
        uint64_t offset = 0, offset2 = 0;
        long snap_ver, snap_from;
        char uuid[MAX_NAME_LEN] = {};
        char snap_name[MAX_NAME_LEN];

        ret = stor_lookup1(pool, name, &chkid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (chkid.type == __POOL_CHUNK__) {
                ret = EISDIR;
                GOTO(err_ret, ret);
        }

        __snapshot_list_init(name);
        get_uuid(uuid);

        ret = stor_snapshot_listopen(&chkid, uuid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        while (done == 0) {
                ret = stor_snapshot_list(&chkid, uuid, offset, buf, &buflen);
                if (unlikely(ret)) {
                        GOTO(err_close, ret);
                }

                if (buflen == 0)
                        break;

                offset2 = 0;
                dir_for_each(buf, buflen, de, offset2) {
                        if (strlen(de->d_name) == 0) {
                                done = 1;
                                break;
                        } else if (buflen - offset2 < sizeof(*de) + MAX_NAME_LEN) {
                                break;
                        }

                        offset += de->d_reclen;

                        if (!memcmp(de->d_name, LICH_SYSTEM_ATTR_UNLINK, strlen(LICH_SYSTEM_ATTR_UNLINK))) {
                                continue;
                        } else if (!memcmp(de->d_name, LICH_SYSTEM_ATTR_AUTO, strlen(LICH_SYSTEM_ATTR_AUTO))) {
                                continue;
                        } else if (!memcmp(de->d_name, LICH_SYSTEM_ATTR_ROOT, strlen(LICH_SYSTEM_ATTR_ROOT))) {
                                continue;
                        } else {
                                sprintf(snap_name, "%s@", name);
                                ret = sscanf(de->d_name, "%s %ld %ld", snap_name+strlen(snap_name), &snap_ver, &snap_from);
                                if (ret != 3)
                                        continue;

                                ret = utils_snapshot_check_protect(pool, snap_name, &protect);
                                if (unlikely(ret)) {
                                        DWARN("check snapshot[%s] protect failed !\n", snap_name);
                                        continue;
                                }

                                if (protect) {
                                        DWARN("There is a protected snapshot on this volume. No deletion is allowed !\n");
                                        ret = EPERM;
                                        GOTO(err_close, ret);
                                }
                        }
                }
        }

        stor_snapshot_listclose(&chkid, uuid);
        return 0;
err_close:
        stor_snapshot_listclose(&chkid, uuid);
err_ret:
        return ret;
}

/**
 * @note 新卷保留原卷的精简配置和hostmap属性
 * @note 目前实现为离线拷贝
 *
 * @param fpool
 * @param from
 * @param tpool
 * @param to
 * @param p
 * @param storage_area
 * @param type
 * @return
 */
int utils_duplicate(const char *fpool, const char *from, const char *tpool,
                const char *to, int p, const char *storage_area, int type)
{
        int ret, ret1, retry = 0;
        char snap_name[MAX_NAME_LEN];
        uuid_t _uuid;
        char uuid[UUID_LEN] = {};
        fileid_t fileid;
        fileinfo_t fileinfo;
        char value[MAX_INFO_LEN];
        int valuelen = MAX_INFO_LEN;

        uuid_generate(_uuid);
        uuid_unparse(_uuid, uuid);

        if (type)
                sprintf(snap_name, "%s@%s", from, LICH_MIGRATE_TAG);
        else
                sprintf(snap_name, "%s@%s", from, uuid);

        /****  1. storage area is valid  *****/
        if (!cluster_storage_area_is_null(storage_area)) {
                ret = dispatch_check_storage_area(storage_area);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        /****  2. create snapshot  ****/
        ret = utils_snapshot_create(fpool, snap_name, TRUE, FALSE);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        /****  3. snapshot clone  ****/
        ret = utils_snapshot_clone(fpool, snap_name, tpool, to, p);
        if (unlikely(ret)) {
                if (ret == EEXIST)
                        GOTO(err_remove_snap, ret);
                else
                        GOTO(err_remove_volume, ret);
        }

        // clone出来的卷默认是精简卷，需要与原卷属性一致
        ret = lich_xattr_get_buf(fpool, from, LICH_SYSTEM_ATTR_THIN, value, &valuelen);
        if (unlikely(ret)) {
                if (ret != ENOKEY)
                        GOTO(err_remove_volume, ret);
        }

        if (strcmp(value, "thick") == 0 && ret == 0) {
                ret = lich_xattr_set(tpool, to, LICH_SYSTEM_ATTR_THIN, value);
                if (unlikely(ret))
                        GOTO(err_remove_volume, ret);
        }

        /****  4. flatten  *****/
        ret = utils_snapshot_flat(tpool, to);
        if (unlikely(ret))
                GOTO(err_remove_volume, ret);

        /****  5. remove snapshot ****/
        ret = stor_lookup1(tpool, to, &fileid);
        if (unlikely(ret)) {
                GOTO(err_remove_volume, ret);
        }

        while (1) {
                ret = md_getattr(&fileid, &fileinfo);
                if (ret) {
                        GOTO(err_remove_volume, ret);
                }

                if (is_row2(&fileinfo) || is_lsv(&fileinfo)) {
                        if (fileinfo.source == 0) {
                                ret = utils_snapshot_remove(fpool, snap_name, 0);
                                if (unlikely(ret)) {
                                        goto out;
                                }
                                break;
                        }
                } else {
                        if (!(fileinfo.attr & __FILE_ATTR_FLAT__)) {
                                ret = utils_snapshot_remove(fpool, snap_name, 0);
                                if (unlikely(ret)) {
                                        goto out;
                                }
                                break;
                        }
                }

                sleep(2);
        }

retry1:
        ret = hostmap_inherit(fpool, from, tpool, to);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_remove_volume, ret, retry1, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                } else
                        GOTO(err_remove_volume, ret);
        }
out:
        return 0;
err_remove_volume:
        retry = 0;
        retry:
        ret1 = utils_rmvol(tpool, to, 0);
        if (unlikely(ret1)) {
                ret1 = _errno(ret1);
                if (ret1 == ENOENT || ret1 == ENOKEY) {
                        // Do nothing
                } else if (ret1 == EAGAIN || ret1 == ENOSPC) {
                        USLEEP_RETRY(err_remove_snap, ret, retry, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                } else
                        DWARN("recycle volume %s/%s failed, ret: %d\n", tpool, to, ret1);
        }
err_remove_snap:
        utils_snapshot_remove(fpool, snap_name, 0);
err_ret:
        return ret;
}

int utils_duplicate_rm_srcvol(const char *pool, const char *file, int force)
{
        int ret, empty = 0;
        char name[MAX_NAME_LEN];
        fileid_t parent, fid;

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, file, &parent, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &parent, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (fid.type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = md_check_connection(&fid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (force) {
                ret = utils_volume_has_protected_snapshot(pool, file);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        } else {
                ret = md_snapshot_isempty(&fid, &empty);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                if (!empty) {
                        ret = EPERM;
                        DWARN("There are snapshots on this volume. No deletion is allowed !\n");
                        GOTO(err_ret, ret);
                }
        }

        /* 和 utils_rmvol 区别就是不会将 force 参数带进删除中*/
        ret = md_rmvol(&parent, name, 0);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = unmap_hosts_by_fileid(id2str(&fid));
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

int utils_duplicate2(const char *fpool, const char *from, const char *tpool,
                     const char *to, int p, const char *storage_area, int type)
{
        (void) fpool;
        (void) from;
        (void) tpool;
        (void) to;
        (void) p;
        (void) storage_area;
        (void) type;

        return 0;
}

int utils_list_storagearea()
{
        int ret, count = MAX_BUF_LEN, left;
        const char *site;
        char buf[MAX_BUF_LEN], zerobuf[MAX_BUF_LEN];
        const void *ptr;

        memset(buf, 0x00, MAX_BUF_LEN);
        memset(zerobuf, 0x00, MAX_BUF_LEN);
        ret = dispatch_list_storage_area(buf, &count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ptr = buf;
        if (memcmp(buf, zerobuf, MAX_BUF_LEN)) {
                while (ptr - (void *)buf < count) {
                        left = count - (ptr - (void *)buf);
                        ptr = _opaque_decode(ptr, left, &site, NULL, NULL);

                        printf("%s\n", site);
                }
        }

        return 0;
err_ret:
        return ret;
}

int utils_check_storagearea(const char * _site)
{
        int ret;

        ret = dispatch_check_storage_area(_site);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int utils_migrate_lock(etcd_lock_t *lock, const char *key)
{
        int ret;

        ret = etcd_lock_init(lock, ETCD_MIGRATE, key, gloconf.rpc_timeout, -1, -1);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = etcd_lock(lock);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

void utils_migrate_unlock(etcd_lock_t *lock)
{
        etcd_lock_delete(lock);
}

int utils_migrate_condition_check(const char *pool, const char *name, int force)
{
        int ret, retry = 0;
        char snap_name[MAX_NAME_LEN];
        fileid_t fid;

        ret = utils_check_connection(pool, name, &fid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        // check snapshots

        ret = md_snapshot_check(&fid, LICH_MIGRATE_TAG);
        if (unlikely(ret)) {
                if (ret == ENOKEY || ret == ENOENT) {
                        // nothing to do
                } else {
                        GOTO(err_ret, ret);
                }
        } else {
retry:
                // UNDO
                sprintf(snap_name, "%s@%s", name, LICH_MIGRATE_TAG);
                ret = utils_snapshot_remove(pool, snap_name, 0);
                if (unlikely(ret)) {
                        if (ret != ENOKEY) {
                                ret = _errno(ret);
                                if (ret == EAGAIN || ret == ENOSPC) {
                                        USLEEP_RETRY(err_ret, ret, retry, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                                } else
                                        GOTO(err_ret, ret);
                        }
                }
        }

        // here, ensure snapshot LICH_MIGRATE_TAG not exists

        if (force) {
                ret = utils_volume_has_protected_snapshot(pool, name);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

        } else {
                ret = utils_check_snapshot_isempty(pool, name);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

/**
 * @pre 无连接，无用户快照
 *
 * @note 目前实现为离线迁移，在线迁移需要考虑IQN的不变性
 * @note 迁移后，卷ID发生了变化
 *
 * @param src_pool
 * @param from
 * @param dest_pool
 * @param to
 * @param p
 * @param storage_area
 * @param force
 * @return
 */
int utils_migrate(const char *src_pool, char *from, const char *dest_pool,
                char *to, int p, char *storage_area, int force)
{
        int ret, multPool = 1;

        /**
         * 1.in the same pool migration, call 'utils_rename'
         * 2.in different pool migration, call 'utils_duplicate' and 'utils_rmvol'
         * Note: the version don't support multiple pool, so set multPool to 0.
         */
        if (strcmp(src_pool, dest_pool))
                multPool = 1;
        else
                multPool = 0;

        if (multPool) {
                ret = utils_migrate_condition_check(src_pool, from, force);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                ret = utils_duplicate(src_pool, from, dest_pool, to, p, storage_area, 1);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                /*
                 * 这里不用utils_rmvol 是因为删除卷不能加force参数
                 * force只删除了pool下的iteam
                 */
                //ret = utils_rmvol(src_pool, from, force);
                ret = utils_duplicate_rm_srcvol(src_pool, from, force);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        } else {
                ret = utils_rename(src_pool, from, to);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

/**
int utils_migrate(const char *path, const char *storage_area)
{
        int ret, skip_count, new_count, new_sub_count, retry = 0, localize = 0, i;
        uint64_t chknum, j;
        fileid_t fileid, chkid;
        chkinfo_t *chkinfo, *subchkinfo;
        fileinfo_t fileinfo;
        char _chkinfo[CHKINFO_MAX], _subchkinfo[CHKINFO_MAX];
        diskid_t new_diskid[LICH_REPLICA_MAX] = {}, new_sub_diskid[LICH_REPLICA_MAX] = {};
        nid_t nid, skip[LICH_REPLICA_MAX*2];

        if (!cluster_storage_area_is_null(storage_area)) {
                ret = dispatch_check_storage_area(storage_area);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if(fileid.type != __VOLUME_CHUNK__){
                printf("Please input valid path: \n    such as: /iscsi/pool/1\n");
                GOTO(err_ret, ret);
        }

        ret = md_getattr(&fileid, &fileinfo);
        if (ret)
                GOTO(err_ret, ret);

        chkinfo = (void *)_chkinfo;
        ret = md_chunk_getinfo(pool, NULL, &fileid, chkinfo, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        //  1. get new disk in storage area
        if ((fileinfo.attr & __FILE_ATTR_LOCALIZE__) ==  __FILE_ATTR_LOCALIZE__) {
                localize = 1;
        }

        new_count = chkinfo->repnum;
        skip_count = chkinfo->repnum;
        ret = dispatch_newdisk(pool, new_diskid, (uint32_t *)&new_count, storage_area, NULL, NULL, 0, 0);
        if (ret) {
                DERROR("dispatch new disk fail, ret:%d, "CHKID_FORMAT"\n", ret, CHKID_ARG(&chkinfo->id));
                GOTO(err_ret, ret);
        }

        if (new_count < skip_count) {
                ret = ENOSPC;
                GOTO(err_ret, ret);
        }

        //  2. set attr STORAGE_AREA_KEY
        ret = md_map_getsrv(&fileid, &nid);
        if (ret)
                GOTO(err_ret, ret);

        ret = md_xattr_set(&nid, &fileid, STORAGE_AREA_KEY, storage_area, strlen(storage_area), 0);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        //  3. move raw chunk
        subchkinfo = (void *)_subchkinfo;
        chknum = size2chknum(fileinfo.size, &fileinfo.ec);
        for (j = 0; j < chknum; j++) {
                fid2cid(&chkid, &fileid, j);
retry:
                ret = md_chunk_getinfo(pool, &fileid, &chkid, subchkinfo, NULL);
                if (unlikely(ret)) {
                        ret = _errno(ret);
                        if (ret == EAGAIN) {
                                sleep(1);
                                SLEEP_RETRY3(err_ret, ret, retry, retry, 100);
                        } else if (ret == ENOENT || ret == ENOKEY) {
                                continue;
                        } else {
                                GOTO(err_ret, ret);
                        }
                }

                if (localize) {
                        skip[0] = new_diskid[0];
                        new_sub_count = subchkinfo->repnum - 1;
                        ret = dispatch_newdisk(pool, new_sub_diskid, (uint32_t *)&new_sub_count, storage_area, NULL, skip, 1, 0);
                        if (ret) {
                                DERROR("dispatch new disk fail, ret:%d, "CHKID_FORMAT"\n", ret, CHKID_ARG(&subchkinfo->id));
                                GOTO(err_ret, ret);
                        }
                        for (i = new_sub_count - 1; i >= 0; i--) {
                                new_sub_diskid[i + 1] = new_sub_diskid[i];
                        }
                        new_sub_diskid[0] = new_diskid[0];
                        new_sub_count = new_sub_count + 1;
                } else {
                        new_sub_count = subchkinfo->repnum;
                        ret = dispatch_newdisk(pool, new_sub_diskid, (uint32_t *)&new_sub_count, storage_area, NULL, NULL, 0, 0);
                        if (ret) {
                                DERROR("dispatch new disk fail, ret:%d, "CHKID_FORMAT"\n", ret, CHKID_ARG(&subchkinfo->id));
                                GOTO(err_ret, ret);
                        }
                }

retry1:
                ret = md_chunk_move(&(chkinfo->diskid[0].id), &fileid, (const chkid_t *)&chkid, (const nid_t *)new_sub_diskid, new_sub_count);
                if (unlikely(ret)) {
                        ret = _errno(ret);
                        if (ret == EAGAIN) {
                                sleep(1);
                                SLEEP_RETRY3(err_ret, ret, retry1, retry, 100);
                        } else {
                                GOTO(err_ret, ret);
                        }
                }
        }

        //  4. move vol chunk
        ret = md_chunk_move(&(chkinfo->diskid[0].id), &(chkinfo->id), (const chkid_t *)&fileid, (const nid_t *)new_diskid, new_count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}
**/


static int __utils_pool_list_print(void *name, void *arg)
{
        int ret;
        char _chkinfo[CHKINFO_MAX];
        chkinfo_t *chkinfo = (void *)_chkinfo;
        char tmp[MAX_BUF_LEN];
        int verbose = *((int *)arg);

        if (verbose) {
                memset(_chkinfo, 0x00, CHKINFO_MAX);
                ret = system_get_pool_info(name, (void *)chkinfo);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                CHKINFO_STR(chkinfo, tmp);

                printf("%s --> %s\n", (char *)name, tmp);
        } else {
                printf("%s\n", (char *)name);
        }

        return 0;
err_ret:
        return ret;
}

int utils_pool_list(int verbose)
{
        return system_pool_iterator(__utils_pool_list_print, (void *)&verbose);
}

int utils_pool_create(const char *pool)
{
        int ret;
        int count = 0;
        char task[MAX_BUF_LEN];

        ret = network_connect_master();
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        const char *prefix = "task/poolremove";

        ret = etcd_get_text(prefix, pool, task, NULL);
        if (unlikely(ret == 0)) {
                ret = EBUSY;
                GOTO(err_ret, ret);
        }

        ret = system_pool_count(&count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (count >= POOL_MAX) {
                printf("Maximum support creates 256 storage pools, preset %d.\n", count);
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        ret = system_pool_add(pool);
        if (unlikely(ret)) {
                if (ret == EEXIST) {
                        //nothing todo;
                } else
                        GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static int __utils_pool_remove_isempty__(void *pool, void *_fid, void *_name, void *_empty)
{
        (void) pool;
        (void) _fid;
        (void) _name;

        *(int *)_empty = 0;

        return 0;
}

static int __utils_pool_remove_isempty(void *pool, void *_fid, void *_name, void *_empty)
{
        int ret;
        char name[MAX_NAME_LEN];

        (void) pool;
        (void) _fid;

        sprintf(name, "/%s", (char *)_name);
        if (strcmp((char *)_name, "system")) {
                ret = utils_iterator(pool, name, __utils_pool_remove_isempty__, _empty);
                if (unlikely(ret)) {
                        if (ret != ENOSPC)
                                GOTO(err_ret, ret);
                }
        }

        return 0;
err_ret:
        return ret;
}

static int __utils_pool_drop(void *node, void *pool)
{
        int ret;

        ret = node_pooldrop(node, pool);
        if (unlikely(ret)) {
                /* nothing todo */
        }

        return 0;
}

int utils_pool_remove(const char *pool)
{
        int ret;
        int empty = 1;

        ret = utils_iterator(pool, "/", __utils_pool_remove_isempty, &empty);
        if (unlikely(ret)) {
                if (ret != ENOSPC)
                        GOTO(err_ret, ret);
        }

        if (!empty) {
                ret = ENOTEMPTY;
                GOTO(err_ret, ret);
        }

        const char *prefix = "task/poolremove";

        ret = etcd_set_with_ttl(prefix, pool, "1", 0);
        if (unlikely(ret)) {
                DWARN("set path %s key %s ret %d\n", prefix, pool, ret);
                GOTO(err_ret, ret);
        }

        // 清除etcd记录
        ret = system_pool_del(pool);
        if (unlikely(ret)) {
                if (ret == EPERM)
                        ret = EAGAIN;
                GOTO(err_ret, ret);
        }

        // 清除各个节点上的相关信息
        ret = cluster_listnode_iterator(__utils_pool_drop, (void *)pool);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = etcd_del(prefix, pool);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        DINFO("pool %s removed\n", pool);
        return 0;
err_ret:
        return ret;
}

int utils_pool_drop(const char *pool)
{
        int ret, find;
        char host[MAX_NAME_LEN];

        ret = system_pool_find(pool, &find);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (find) {
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        ret = net_gethostname(host, MAX_MSG_SIZE);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = node_pooldrop(host, pool);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

static void __utils_allcoate_chkid(const fileid_t *fileid,
                int begin, int end, chkid_t *chkid)
{
        int i, idx;

        i = 0;
        for (idx = begin; idx < end; idx++) {
                fid2cid(&chkid[i++], fileid, idx);
        }
}

static int __utils_allcoate_pipeline(allocate_arg_t *arg)
{
        int ret, retry, begin, end, pipeline, user_pipeline, fill, left, step;
        const fileid_t *fileid;
        chkid_t *chkid;
        struct timeval t1, t2;
        int64_t used;

        fileid = &arg->fileid;
        begin = arg->begin;
        end = arg->end;
        fill = arg->fill;
        user_pipeline = arg->user_pipeline;
        pipeline = arg->pipeline;

        // slow start，最大值是user_pipeline

        ret = ymalloc((void **)&chkid, sizeof(*chkid)*user_pipeline);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        left = end - begin;
        while (left > 0) {
                retry = 0;
retry:
                step = left < pipeline ? left : pipeline;
                if (step > user_pipeline)
                        step = user_pipeline;

                YASSERT(step > 0);

                __utils_allcoate_chkid(fileid, begin, begin+step, chkid);

                _gettimeofday(&t1, NULL);

                ret = md_chunk_allocate(fileid, chkid, step, fill, NULL);
                {
                        _gettimeofday(&t2, NULL);
                        used = _time_used(&t1, &t2);

                        arg->used = used;

                        // printf("=== begin %u step %u used %ju ret %d\n", begin, step, used, ret);
                }
                if (unlikely(ret)) {
                        if (ret == ECANCELED)
                                GOTO(err_free, ret);

                        ret = _errno(ret);
                        if (ret == ETIMEDOUT || ret == EAGAIN) {
                                if (pipeline > 10) {
                                        pipeline /= 2;
                                }
                                arg->pipeline = pipeline;
                                USLEEP_RETRY(err_free, ret, retry, retry, gloconf.rpc_timeout * 4, (1000 * 1000));
                        } else
                                GOTO(err_free, ret);
                } else {
                        if (used < (gloconf.rpc_timeout / 4)  * 1000 * 1000
                                && pipeline < user_pipeline) {
                                pipeline += 10;
                                arg->pipeline = pipeline;
                        }
                }

                begin += step;
                left -= step;
                arg->success += step;
        }

        yfree((void **)&chkid);

        return 0;
err_free:
        yfree((void **)&chkid);
err_ret:
        return ret;
}

void * __lich_allocate_thread(void *_arg)
{
        int ret;
        allocate_arg_t *arg;
        //chkinfo_t *chkinfo;
        //char buf[MAX_BUF_LEN], str[MAX_BUF_LEN];

        arg = _arg;
        ret = __utils_allcoate_pipeline(arg);
        if (ret) {
                GOTO(err_ret, ret);
        }

        arg->ret = 0;
        return NULL;
err_ret:
        arg->ret = ret;
        return NULL;
}

static uint64_t success_pre;

static void * __lich_allocate_speed(const fileid_t *fileid,
                uint64_t success, uint64_t total, uint64_t pipeline, int64_t used)
{
        int speed, finish;

        finish = success - success_pre;
        speed = finish / ((double)used / (1000 * 1000));

        if (finish && success_pre)
                printf("allocate %s, success: %ju, total: %ju, pipeline %ju used %fs speed: %d chunk/s\n",
                                id2str(fileid), success, total, pipeline, (double)used / (1000 * 1000), speed);

        success_pre = success;

        return NULL;
}

static void * __lich_allocate_count(void *_arg)
{
        int ret, i;
        allocate_arg_count_t *count = _arg;
        allocate_arg_t *arg;
        uint64_t success = 0, pipeline = 0;
        int64_t used;

        success_pre = 0;
        while (count->running) {
                success = 0;
                pipeline = 0;
                used = 0;
                for (i = 0; i < count->thread; i++) {
                        arg = &count->arg[i];
                        success += arg->success;
                        pipeline += arg->pipeline;
                        if (used < arg->used)
                                used = arg->used;
                }

                ret = _sem_timedwait1(&count->sem, 2);
                if (unlikely(ret)) {
                        if (ret == ETIMEDOUT) {
                        } else
                                UNIMPLEMENTED(__WARN__);
                }

                __lich_allocate_speed(&arg->fileid, success, count->total, pipeline, used);
        }

        if (success != count->total) {
                success = 0;
                pipeline = 0;
                used = 0;
                for (i = 0; i < count->thread; i++) {
                        arg = &count->arg[i];
                        success += arg->success;
                        pipeline += arg->pipeline;
                        if (used < arg->used)
                                used = arg->used;
                }

                __lich_allocate_speed(&arg->fileid, success, count->total, pipeline, used);
                //YASSERT(success == count->total);
        } else {
                __lich_allocate_speed(&arg->fileid, success, count->total, pipeline, used);
                //YASSERT(success == count->total);
        }

        sem_post(&count->stop);

        return NULL;
}

int utils_allocate(const char *pool, const char *path, int pipeline, int concurrent, int fill)
{
        int ret, chknum, left, avg_size, step, i, count, begin;
        fileid_t fileid;
        fileinfo_t fileinfo;
        allocate_arg_t *args;
        allocate_arg_count_t _count;
        pthread_t th;

        ret = ymalloc((void **)&args, sizeof(*args) * ALLOCATE_THREAD_MAX);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_free, ret);

        ret = md_getattr(&fileid, &fileinfo);
        if (unlikely(ret)) {
                GOTO(err_free, ret);
        }

        concurrent = _min(concurrent, ALLOCATE_THREAD_MAX);
        pipeline = _min(pipeline, ALLOCATE_PIPE_MAX);
        if (EC_ISEC(&fileinfo.ec)) {
                pipeline = 1;
        }
        printf("allocate %s pipeline:%d concurrent: %d fill:%d\n",
                        path, pipeline, concurrent, fill);

        chknum = size2chknum(fileinfo.size, &fileinfo.ec);
        left = chknum;
        avg_size = _ceil(left, concurrent);
        count = 0;
        begin = 0;
        while (left > 0) {
                step = left < avg_size ? left : avg_size;
                args[count].fileid = fileid;
                args[count].begin = begin;
                args[count].end = begin + step;
                args[count].fill = fill;
                args[count].ret = 0;
                args[count].success = 0;
                args[count].user_pipeline = pipeline;
                args[count].pipeline = 10;
                args[count].used = 0;
                strcpy(args[count].pool, pool);

                count++;
                begin += step;
                left -= step;
        }

        for (i=0; i < count; i++) {
                ret = pthread_create(&args[i].th, NULL, __lich_allocate_thread, &args[i]);
                if (unlikely(ret))
                        GOTO(err_free, ret);
        }

        _count.running = 1;
        _count.total = chknum;
        _count.arg = args;
        _count.thread = count;

        ret = sem_init(&_count.sem, 0, 0);
        if (unlikely(ret))
                GOTO(err_free, ret);

        ret = sem_init(&_count.stop, 0, 0);
        if (unlikely(ret))
                GOTO(err_free, ret);

        ret = pthread_create(&th, NULL, __lich_allocate_count, &_count);
        if (unlikely(ret))
                GOTO(err_free, ret);


        for (i=0; i < count; i++) {
                pthread_join(args[i].th, NULL);
        }

        _count.running = 0;
        sem_post(&_count.sem);

        ret = sem_wait(&_count.stop);
        if (unlikely(ret))
                GOTO(err_free, ret);

        for (i=0; i < count; i++) {
                ret = args[i].ret;
                if (unlikely(ret)) {
                        GOTO(err_free, ret);
                }
        }

        yfree((void **)&args);
        return 0;
err_free:
        yfree((void **)&args);
err_ret:
        return ret;
}

int utils_md5sum(const char *pool, const char *path, int mult_thread)
{
        int ret;
        fileid_t fileid;

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (mult_thread)
                ret = file_md5sum_mult_thread(pool, &fileid, path);
        else
                ret = file_md5sum(pool, &fileid, path);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int utils_connection(const char *pool, const char *path)
{
        int ret, count = MAX_BUF_LEN, i, left;
        fileid_t fileid;
        chkinfo_t *chkinfo;
        char _chkinfo[CHKINFO_MAX];
        char _buf[MAX_BUF_LEN], tmp[MAX_BUF_LEN], buf[MAX_BUF_LEN];
        const nid_t *nid;
        const uint32_t *addrs;
        uint32_t size;
        const void *ptr;

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if(fileid.type != __VOLUME_CHUNK__){
                printf("Please input valid path: \n    such as: /iscsi/pool/1\n");
                GOTO(err_ret, ret);
        }

        chkinfo = (void *)_chkinfo;
        ret = md_chunk_getinfo(pool, NULL, &fileid, chkinfo, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        memset(buf, 0x00, MAX_BUF_LEN);
        memset(_buf, 0x00, MAX_BUF_LEN);
        ret = md_connection(&chkinfo->diskid[0].id, &fileid, _buf, &count);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (memcmp(_buf, buf, MAX_BUF_LEN)) {
                buf[0] = '\0';
                ptr = _buf;
                while (ptr - (void *)_buf < count) {
                        left = count - (ptr - (void *)_buf);
                        ptr = _opaque_decode(ptr, left, &nid, NULL, &addrs, &size, NULL);

                        tmp[0] = '\0';
                        for (i = 0; i < (int)(size / sizeof(*addrs)); i++) {
                                if (i == 0)
                                        snprintf(tmp, MAX_BUF_LEN, "%s", _inet_ntoa(addrs[i]));
                                else
                                        snprintf(tmp + strlen(tmp), MAX_BUF_LEN, ",%s", _inet_ntoa(addrs[i]));
                        }

                        snprintf(buf + strlen(buf), MAX_BUF_LEN, "%s:%s\n", netable_rname_nid(nid), tmp);
                }
        }

        printf("%s\n", buf);

        return 0;
err_ret:
        return ret;
}

int utils_check_connection(const char *pool, const char *file, fileid_t *fid)
{
        int ret;
        char name[MAX_NAME_LEN];
        fileid_t parent;

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, file, &parent, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &parent, name, fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (fid->type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = md_check_connection(fid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        return 0;
err_ret:
        return ret;
}

int utils_check_snapshot_isempty(const char *pool, const char *file)
{
        int ret, empty = 0;
        char name[MAX_NAME_LEN];
        fileid_t parent, fid;

        if (!pool) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = stor_splitpath(pool, file, &parent, name);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        ret = stor_lookup(pool, &parent, name, &fid);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        if (fid.type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }

        ret = md_snapshot_isempty(&fid, &empty);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (!empty) {
                ret = EPERM;
                DWARN("There are snapshots on this volume. No deletion is allowed !\n");
                GOTO(err_ret, ret);
        }

        return 0;
err_ret:
        return ret;
}

void * __stor_stat_thread(void *_arg)
{
        int ret;
        filestat_t _filestat, *filestat;
        uint64_t left, step, begin;
        stor_stat_t *arg = _arg;

        DBUG("stat_thread idx %d "CHKID_FORMAT" off %ju size %ju\n", arg->idx, CHKID_ARG(&arg->fileid), arg->off, arg->size);

        filestat = &arg->filestat;
        filestat->sparse = 0;
        filestat->localized = 0;
        filestat->rollback = 0;
        left = arg->size;
        begin = arg->off;

        while (left > 0) {
                step = left < STOR_STAT_STEP_MAX ? left : STOR_STAT_STEP_MAX;
                ret = stor_rpc_stat(&arg->nid, &arg->fileid, &_filestat, begin, step);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                begin += step;
                left -= step;
                filestat->sparse += _filestat.sparse;
                filestat->localized += _filestat.localized;
                filestat->rollback += _filestat.rollback;
        }


        DBUG("stat_thread idx %d "CHKID_FORMAT" off %ju size %ju res %d %d\n", arg->idx, CHKID_ARG(&arg->fileid),
                        arg->off, arg->size, arg->filestat.sparse, arg->filestat.localized);
        arg->ret = 0;
        return NULL;
err_ret:
        arg->ret = ret;
        return NULL;

}

int stor_stat_multi(const nid_t *nid, const fileid_t *fileid, filestat_t *filestat, int multi, int full)
{
        int thread, len, ret;
        uint64_t i, chknum, left, step, count, begin, avg_size;
        fileinfo_t fileinfo;
        stor_stat_t args[STOR_STAT_THREAD_MAX];
        pthread_t ths[STOR_STAT_THREAD_MAX];

        DBUG("multi %d full %d\n", multi, full);

        ret = md_getattr(fileid, &fileinfo);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

        filestat->size = fileinfo.size;
        filestat->snap_version = fileinfo.snap_version;
        filestat->snap_rollback = fileinfo.snap_rollback;
        filestat->snap_from = fileinfo.snap_from;
        filestat->attr = fileinfo.attr;

        chknum = size2chknum(fileinfo.size, &fileinfo.ec);
        filestat->chknum = chknum;
        filestat->repnum = fileinfo.repnum_usr;

        filestat->sparse = 0;
        filestat->localized = 0;
        filestat->rollback = 0;

        if (filestat->attr & __FILE_ATTR_CLONE__) {
                len = MAX_NAME_LEN;
                ret = md_xattr_get(nid, fileid, LICH_SYSTEM_ATTR_SOURCE, filestat->src, &len);
                if (unlikely(ret))
                        GOTO(err_ret, ret);
        }

        if (full == VOL_INFO_NORMAL) {
                // nothing to do
        } else if (full == VOL_INFO_ALLOCATE){
                filestat_t _filestat;

                ret = stor_rpc_stat(nid, fileid, &_filestat, 0, 0);
                if (unlikely(ret))
                        GOTO(err_ret, ret);

                // sparse即是分配的chknum
                filestat->sparse = _filestat.sparse;
        } else if (full == VOL_INFO_FULL){
                if (!multi || chknum < STOR_STAT_THREAD_MAX)
                        thread = 1;
                else
                        thread = STOR_STAT_THREAD_MAX;

                left = chknum;
                avg_size = _ceil(left, thread);
                count = 0;
                begin = 0;
                while (left > 0) {
                        args[count].idx = count;
                        step = left < avg_size ? left : avg_size;
                        args[count].nid = *nid;
                        args[count].fileid = *fileid;
                        args[count].off = begin;
                        args[count].size = step;

                        count++;
                        begin+=step;
                        left-=step;
                }

                for (i=0; i < count; i++) {
                        ret = pthread_create(&ths[i], NULL, __stor_stat_thread, &args[i]);
                        if (unlikely(ret))
                                GOTO(err_ret, ret);
                }

                for (i=0; i < count; i++) {
                        pthread_join(ths[i], NULL);
                }

                for (i=0; i < count; i++) {
                        ret = args[i].ret;
                        if (unlikely(ret)) {
                                GOTO(err_ret, ret);
                        } else {
                                filestat->sparse += args[i].filestat.sparse;
                                filestat->localized += args[i].filestat.localized;
                                filestat->rollback +=args[i].filestat.rollback;
                        }
                }
        }

        DBUG("chknum %d sparse %d localized %d rollback %d\n",
             filestat->chknum, filestat->sparse, filestat->localized, filestat->rollback);

        return 0;
err_ret:
        return ret;
}

void utils_info_print(fileid_t *fileid, nid_t *nid, filestat_t *filestat, int full)
{
        printf("chkid : "CHKID_FORMAT"\n"
               "location : %s\n"
               "size : %ju\n"
               "chknum : %u\n"
               "repnum : %u\n"
               "snap_from : %jd\n"
               "snap_version : %jd\n"
               "snap_rollback : %jd\n",
               CHKID_ARG(fileid),
               network_rname(nid),
               filestat->size,
               filestat->chknum,
               filestat->repnum,
               filestat->snap_from,
               filestat->snap_version,
               filestat->snap_rollback);

        if (full == VOL_INFO_ALLOCATE) {
                YASSERT(filestat->sparse <= filestat->chknum);

                printf("allocated : %u\n", filestat->sparse);
        } else if (full == VOL_INFO_FULL) {
                YASSERT(filestat->sparse <= filestat->chknum);

                printf("allocated : %u\n"
                       "localized : %u\n"
                       "rollback : %u\n",
                       filestat->chknum - filestat->sparse,
                       filestat->localized,
                       filestat->rollback);
        }

        if (filestat->attr & __FILE_ATTR_LSV__) {
                printf("logical_size: %ju\n", filestat->logical_size);
        }

        if (filestat->attr & __FILE_ATTR_SNAPSHOT__) {
                printf("snapshot : yes\n");
                printf("protected : %s\n", ((filestat->attr & __FILE_ATTR_PROTECT__) ? "yes" : "no"));
        }
        if (filestat->attr & __FILE_ATTR_CLONE__) {
                printf("source : %s\n", filestat->src);
                printf("flat : %s\n", (filestat->attr & __FILE_ATTR_FLAT__ ? "on" : "off"));
        }

        if (filestat->attr & __FILE_ATTR_LSV__) {
                printf("* format : lsv\n");
                printf("* inited : %d\n", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                printf("* volume_page_id : %u\n", filestat->volume_page_id);
                printf("* gc_os_page_id : %u\n", filestat->gc_os_page_id);
                printf("* gc_bitmap_page_id : %u\n", filestat->gc_bitmap_page_id);
                printf("* rcache_page_id : %u\n", filestat->rcache_page_id);
                printf("* wbuf_page_id : %u\n", filestat->wbuf_page_id);
                printf("* bitmap_chunk_id : %u\n", filestat->bitmap_chunk_id);
                printf("* source : %ju\n", filestat->source);
                printf("* snap : %s\n", filestat->snap);
        } else if (filestat->attr & __FILE_ATTR_ROW2__) {
                printf("* format : row2\n");
                printf("* inited : %d\n", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                printf("* volume_page_id : %u\n", filestat->volume_page_id);
                //printf("* gc_os_page_id : %u\n", filestat->gc_os_page_id);
                //printf("* gc_bitmap_page_id : %u\n", filestat->gc_bitmap_page_id);
                //printf("* rcache_page_id : %u\n", filestat->rcache_page_id);
                //printf("* wbuf_page_id : %u\n", filestat->wbuf_page_id);
                printf("* bitmap_chunk_id : %u\n", filestat->bitmap_chunk_id);
                printf("* source : %ju\n", filestat->source);
                printf("* snap : %s\n", filestat->snap);
        } else if (filestat->attr & __FILE_ATTR_ROW3__) {
                printf("* format : row3\n");
                printf("* inited : %d\n", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                printf("* volume_page_id : %u\n", filestat->volume_page_id);
                //printf("* gc_os_page_id : %u\n", filestat->gc_os_page_id);
                //printf("* gc_bitmap_page_id : %u\n", filestat->gc_bitmap_page_id);
                //printf("* rcache_page_id : %u\n", filestat->rcache_page_id);
                //printf("* wbuf_page_id : %u\n", filestat->wbuf_page_id);
                printf("* bitmap_chunk_id : %u\n", filestat->bitmap_chunk_id);
                printf("* source : %ju\n", filestat->source);
                printf("* snap : %s\n", filestat->snap);
        }
}

static void utils_info_print_json(fileid_t *fileid, nid_t *nid, filestat_t *filestat, int full)
{
        char chkid[MAX_NAME_LEN];

        //创建一个空的文档(对象)
        cJSON *json = cJSON_CreateObject();

        sprintf(chkid, CHKID_FORMAT, CHKID_ARG(fileid));
        cJSON_AddStringToObject(json, "chkid", chkid);
        cJSON_AddStringToObject(json, "location", network_rname(nid));
        cJSON_AddNumberToObject(json, "size", filestat->size);
#if LSV
        cJSON_AddNumberToObject(json, "logical_size", filestat->logical_size);
#endif
        cJSON_AddNumberToObject(json, "chknum", filestat->chknum);
        cJSON_AddNumberToObject(json, "repnum", filestat->repnum);
        cJSON_AddNumberToObject(json, "snap_version", filestat->snap_version);
        cJSON_AddNumberToObject(json, "snap_rollback", filestat->snap_rollback);

        if (full == VOL_INFO_ALLOCATE) {
                YASSERT(filestat->sparse <= filestat->chknum);

                cJSON_AddNumberToObject(json, "allocated", filestat->sparse);
        } else if (full == VOL_INFO_FULL) {
                YASSERT(filestat->sparse <= filestat->chknum);

                cJSON_AddNumberToObject(json, "allocated", filestat->chknum - filestat->sparse);
                cJSON_AddNumberToObject(json, "localized", filestat->localized);
                cJSON_AddNumberToObject(json, "rollback", filestat->rollback);
        }
        if (filestat->attr & __FILE_ATTR_SNAPSHOT__) {
                cJSON_AddStringToObject(json, "snapshot", "yes");
                cJSON_AddStringToObject(json, "protected",
                        ((filestat->attr & __FILE_ATTR_PROTECT__) ? "yes" : "no"));
        }
        if (filestat->attr & __FILE_ATTR_CLONE__) {
                cJSON_AddStringToObject(json, "source", filestat->src);
                cJSON_AddStringToObject(json, "flat",
                        (filestat->attr & __FILE_ATTR_FLAT__ ? "on" : "off"));
        }

#if LSV
        if (filestat->attr & __FILE_ATTR_LSV__) {
                cJSON_AddStringToObject(json, "* format", "lsv");
                cJSON_AddNumberToObject(json, "* inited", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                cJSON_AddNumberToObject(json, "* volume_page_id", filestat->volume_page_id);
                cJSON_AddNumberToObject(json, "* gc_os_page_id", filestat->gc_os_page_id);
                cJSON_AddNumberToObject(json, "* gc_bitmap_page_id", filestat->gc_bitmap_page_id);
                cJSON_AddNumberToObject(json, "* rcache_page_id", filestat->rcache_page_id);
                cJSON_AddNumberToObject(json, "* wbuf_page_id", filestat->wbuf_page_id);
                cJSON_AddNumberToObject(json, "* bitmap_chunk_id", filestat->bitmap_chunk_id);
                cJSON_AddNumberToObject(json, "* source", filestat->source);
                cJSON_AddStringToObject(json, "* snap", filestat->snap);
        }
        if (filestat->attr & __FILE_ATTR_ROW2__) {
                cJSON_AddStringToObject(json, "* format", "row2");
                cJSON_AddNumberToObject(json, "* inited", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                cJSON_AddNumberToObject(json, "* volume_page_id", filestat->volume_page_id);
                //cJSON_AddNumberToObject(json, "* gc_os_page_id", filestat->gc_os_page_id);
                //cJSON_AddNumberToObject(json, "* gc_bitmap_page_id", filestat->gc_bitmap_page_id);
                //cJSON_AddNumberToObject(json, "* rcache_page_id", filestat->rcache_page_id);
                //cJSON_AddNumberToObject(json, "* wbuf_page_id", filestat->wbuf_page_id);
                cJSON_AddNumberToObject(json, "* bitmap_chunk_id", filestat->bitmap_chunk_id);
                cJSON_AddNumberToObject(json, "* source", filestat->source);
                cJSON_AddStringToObject(json, "* snap", filestat->snap);
        }
        if (filestat->attr & __FILE_ATTR_ROW3__) {
                cJSON_AddStringToObject(json, "* format", "row3");
                cJSON_AddNumberToObject(json, "* inited", fileattr_is_set(filestat->attr, __FILE_ATTR_INITED__));
                cJSON_AddNumberToObject(json, "* volume_page_id", filestat->volume_page_id);
                //cJSON_AddNumberToObject(json, "* gc_os_page_id", filestat->gc_os_page_id);
                //cJSON_AddNumberToObject(json, "* gc_bitmap_page_id", filestat->gc_bitmap_page_id);
                //cJSON_AddNumberToObject(json, "* rcache_page_id", filestat->rcache_page_id);
                //cJSON_AddNumberToObject(json, "* wbuf_page_id", filestat->wbuf_page_id);
                cJSON_AddNumberToObject(json, "* bitmap_chunk_id", filestat->bitmap_chunk_id);
                cJSON_AddNumberToObject(json, "* source", filestat->source);
                cJSON_AddStringToObject(json, "* snap", filestat->snap);
        }

#endif

        printf("%s\n", cJSON_PrintUnformatted(json));

        //释放json结构所占用的内存
        cJSON_Delete(json);
}


/**
 * @todo support snapshot
 *
 * @param pool
 * @param path
 * @param verbose
 * @return
 */

int utils_info(const char *pool, const char *path, int full, int output_format)
{
        int ret, multi = 1;
        fileid_t fileid;
        chkinfo_t *chkinfo;
        char _chkinfo[CHKINFO_MAX];
        filestat_t filestat;

        ret = md_chunk_getid(pool, path, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        chkinfo = (void *)_chkinfo;
        ret = md_chunk_getinfo(pool, NULL, &fileid, chkinfo, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        ret = stor_stat_multi(CHKINFO_FIRSTNID(chkinfo), &fileid, &filestat, multi, full);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        switch(output_format) {
        case DEFAULT_FORMAT:
                utils_info_print(&fileid, CHKINFO_FIRSTNID(chkinfo), &filestat, full);
                break;
        case JSON_FORMAT:
                utils_info_print_json(&fileid, CHKINFO_FIRSTNID(chkinfo), &filestat, full);
                break;
        default:
                fprintf(stderr, "Don't support output format.\n");
                break;
        }

        return 0;
err_ret:
        return ret;
}

int utils_chunkinfo(const char *pool, const char *path, int verbose, int pithy)
{
        int ret;
        chkid_t id;
        char buf[MAX_BUF_LEN];

        verbose = 1;

        ret = md_chunk_getid(pool, path, &id);
        if (unlikely(ret)) {
                GOTO(err_print, ret);
        }

        ret = stor_stat(pool, &id, buf, verbose);
        if (unlikely(ret)) {
                GOTO(err_print, ret);
        }

        printf("%s", buf);

        if (verbose && pithy && id.type == __VOLUME_CHUNK__) {
                ret = stor_stat_file(pool, &id);
                if (unlikely(ret)) {
                        GOTO(err_ret, ret);
                }
        }

        return 0;
err_print:
        if (ret == EAGAIN)
                printf("\x1b[1;31m  *%s : offline\x1b[0m\n", path);
err_ret:
        return ret;
}


int __multpath_get(const char *pool, const char *name, int *multpath)
{
        int ret, retry;
        fileid_t fileid;
        fileinfo_t fileinfo;

        ret = md_chunk_getid(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

#if 0
        if (fileid.type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }
#endif

        retry = 0;
retry:
        ret = md_getattr(&fileid, &fileinfo);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if ((fileinfo.attr & __FILE_ATTR_MULTPATH__) ==  __FILE_ATTR_MULTPATH__) {
                *multpath = 1;
        } else {
                *multpath = 0;
        }


        return 0;
err_ret:
        return ret;
}

int lich_multpath_get(const char *pool, const char *name)
{
        int ret, num;
        ret = __multpath_get(pool, name, &num);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        printf("path: %s, multipath: %d\n", name, num);
        return 0;
err_ret:
        return ret;

}

int lich_multpath_set(const char *pool, const char *name, int num)
{
        int ret, multpath;
        fileid_t fileid;
        setattr_t setattr;

        ret = __multpath_get(pool, name, &multpath);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (multpath == num) {
                goto out;
        }

        ret = md_chunk_getid(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

#if 0
        if (fileid.type != __VOLUME_CHUNK__) {
                ret = EINVAL;
                GOTO(err_ret, ret);
        }
#endif

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.multpath.set_it = 1;
        setattr.multpath.val = num;
        ret = md_setattr(&fileid, NULL, &setattr);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

out:
        return 0;
err_ret:
        return ret;
}

static int __attr_get(const char *pool, const char *name, int flag, int *localize, int *isec)
{
        int ret, retry;
        fileid_t fileid;
        fileinfo_t fileinfo;

        ret = md_chunk_getid(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        retry = 0;
retry:
        ret = md_getattr(&fileid, &fileinfo);
        if (unlikely(ret)) {
                if (ret == EAGAIN) {
                        USLEEP_RETRY(err_ret, ret, retry, retry, 50, (100 * 1000));
                } else
                        GOTO(err_ret, ret);
        }

        if ((fileinfo.attr & flag) ==  flag) {
                *localize = 1;
        } else {
                *localize = 0;
        }

        if (isec) {
                *isec = EC_ISEC(&fileinfo.ec);
        }

        return 0;
err_ret:
        return ret;
}

int lich_readraw_get(const char *pool, const char *name)
{
        int ret, num;

        ret = __attr_get(pool, name, __FILE_ATTR_READRAW__, &num, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        printf("path: %s, readraw: %d\n", name, num);
return 0;
        err_ret:
return ret;
}

int lich_readraw_set(const char *pool, const char *name, int num)
{
        int ret, localize;
        fileid_t fileid;
        setattr_t setattr;

        ret = __attr_get(pool, name, __FILE_ATTR_READRAW__, &localize, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (localize == num) {
                goto out;
        }

        ret = md_chunk_getid(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.readraw.set_it = 1;
        setattr.readraw.val = num;
        ret = md_setattr(&fileid, NULL, &setattr);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

out:
        return 0;
err_ret:
        return ret;
}

int lich_localize_get(const char *pool, const char *name)
{
        int ret, num;

        ret = __attr_get(pool, name, __FILE_ATTR_LOCALIZE__, &num, NULL);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        printf("path: %s, localize: %d\n", name, num);
        return 0;
err_ret:
        return ret;
}

int lich_localize_set(const char *pool, const char *name, int num)
{
        int ret, localize, isec;
        fileid_t fileid;
        setattr_t setattr;

        ret = __attr_get(pool, name, __FILE_ATTR_LOCALIZE__, &localize, &isec);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (localize == num) {
                goto out;
        }

        ret = md_chunk_getid(pool, name, &fileid);
        if (unlikely(ret))
                GOTO(err_ret, ret);

        if (fileid.type == __VOLUME_CHUNK__ && isec) {
                ret = EPERM;
                GOTO(err_ret, ret);
        }

        memset(&setattr, 0x0, sizeof(setattr));
        setattr.localize.set_it = 1;
        setattr.localize.val = num;
        ret = md_setattr(&fileid, NULL, &setattr);
        if (unlikely(ret)) {
                GOTO(err_ret, ret);
        }

out:
        return 0;
err_ret:
        return ret;
}
